background jobs

import "../pg"
import "std/io"
import "std/json"
import "std/list"
import "std/text"
import "std/time"

# A job as its row holds it.
type queued_job
  id
  name
  args
  attempts

# Work an app does after the request that asked for it, as Rails' Active Job
# runs it on Solid Queue: each job is a row of `senro_jobs`, so it outlives
# the process that queued it, and any number of workers can share a queue.
# A job has a queue, the name of the performer that does it, its arguments
# as a map of strings, and the moment from which it may run.
#
# A worker claims one due job at a time with `for update skip locked`, so two
# workers never take the same row, and leases it for five minutes: a worker
# that dies mid-job leaves the row to be claimed again once the lease runs
# out. A job that succeeds is deleted. One whose performer fails is tried
# again later, attempts^4 + 2 seconds after the failure, as Rails'
# `polynomially_longer` waits, and after five attempts it stays in the table
# with the reason, for someone to read.
#
# `prepared` makes the table jobs live in when it is missing.
pub fn prepared db
  pg/run db ledger .> (_ -> effect db)

ledger = """
  create table if not exists senro_jobs (
    id bigserial primary key,
    queue text not null,
    name text not null,
    args text not null,
    run_at bigint not null,
    attempts int not null default 0,
    failure text not null default ''
  )
"""

# A job as its row holds it.
type queued_job
  id
  name
  args
  attempts

# Queues the job `name` with `args` on `queue`, to run at `t` (milliseconds
# since the epoch) or after.
pub fn later db queue name args t
  given = [queue name (json/encode args) "{t}"]
  pg/query db queued given .> (_ -> effect db)

queued = """
  insert into senro_jobs (queue, name, args, run_at)
  values ($1, $2, $3, $4::bigint)
"""

# The attempts a job gets, and how long a claim holds it.
pub attempts_allowed = 5

lease_ms = 5 * 60 * 1000

# Claims one job on `queue` that is due at `t`, and runs it. `performers` maps
# a job's name to a function of the session and the job's arguments that
# answers an effect. The answer says what happened: `none` when no job was
# due, or a line naming the job and how it went.
pub fn worked db queue performers t
  given = [queue "{t}" "{t + lease_ms}" "{attempts_allowed}"]
  pg/query db claimed given .> (r -> ran db performers t (list/first r.rows))

claimed = """
  update senro_jobs set run_at = $3::bigint, attempts = attempts + 1
  where id = (
    select id from senro_jobs
    where queue = $1 and run_at <= $2::bigint and attempts < $4::int
    order by run_at, id
    for update skip locked
    limit 1
  )
  returning id, name, args, attempts
"""

fn ran _ _ _ none
  effect none

# A performer's failure was born in the app, so the worker may rescue it.
# A job whose name has no performer is the worker's own finding, and is
# recorded the same way without being raised.
fn ran db performers t row
  j = job_of row
  run_by db performers[j.name] j t

fn run_by db none j t
  failed db j t "no performer is named {j.name}"

fn run_by db perform j t
  perform db j.args
    .> (_ -> pg/query db finished [j.id])
    .> (_ -> effect "ran {j.name}")
    .? (u -> failed db j t u.reason)

finished = "delete from senro_jobs where id = $1::bigint"

# A failed attempt is kept with its reason, and the job waits attempts^4 + 2
# seconds before it may run again.
fn failed db j t reason
  n = j.attempts
  wait = (n * n * n * n + 2) * 1000
  given = [j.id reason "{t + wait}"]
  said = "failed {j.name} on attempt {n}: {reason}"
  pg/query db kept given .> (_ -> effect said)

kept = """
  update senro_jobs set failure = $2, run_at = $3::bigint
  where id = $1::bigint
"""

import "../../../jobs"
import "std/list"

# The moment the spec starts from, and the moments it works the queue at:
# four at once, then three and sixty seconds on, then pairs far enough apart
# that every job still failing gets an attempt in each, until none has any
# left.
pub t = 1000000

soon = [t t t t (t + 3000) (t + 60000) (t + 60000) (t + 60000)]

apart = [1000000 1000000 10000000 10000000 100000000 100000000]

pub times = list/concat soon (list/to_list (list/map apart (d -> t + d)))

# What each job named on the default queue does: `greet` prints, and
# `broken` fails every time.
pub performers = { "broken":broken "greet":greeted }

fn greeted _ args
  print "hello, {args["name"]}"

fn broken _ _
  err "it broke"

# Works the default queue once at each time in `times`, printing what each
# step did.
pub fn stepped db moments
  list/fold moments (effect db) (acc at -> acc .> (_ -> step db at))

fn step db at
  jobs/worked db "default" performers at
    .> (said -> print "at {at - t}: {told said}")
    .> (_ -> db)

fn told none
  "nothing due"

fn told said
  said

# Works `queue` for as long as the process runs: a job when one is due, and
# a second's sleep when none is. Each job's outcome goes to standard error.
pub fn worker db queue performers
  time/now
    .> (t -> worked db queue performers t)
    .> (said -> told said)
    .> (_ -> worker db queue performers)

# Runs `served`, such as `web/serve`, with a worker on `queue` beside it in a
# fiber of its own, on a session `connect` opens. The worker makes the jobs
# table when it is missing before it starts.
pub fn beside connect queue performers served
  connect .> (db -> prepared db) .> (db -> worker db queue performers)
  served

fn told none
  time/sleep 1000

fn told said
  io/write_err "senro jobs: {said}\n"

# Queues `m` on the `mail` queue of the jobs table, so the request that asked
# for it answers without waiting on the mail server, as Action Mailer's
# `deliver_later` does. A worker running `mail/performers` sends it the way
# `delivered` does. A message `delivered` would refuse is refused here, before
# it is queued.
pub fn delivered_later db m
  why = refusal m
  return err why if why != ""
  fields = { "body":m.body "from":m.from "subject":m.subject "to":m.to }
  time/now .> (t -> jobs/later db "mail" "deliver" fields t)

# The performers a worker on the `mail` queue runs.
pub performers = { "deliver":deliver }

fn deliver _ f
  delivered (message f["from"] f["to"] f["subject"] f["body"])

# The guestbook's mail, as a Rails mailer holds it: what each letter says,
# and the signed link it carries, which starts with `mail/url_for`. Each
# letter is queued on `db` and sent by the worker beside the server. First, a
# link to choose a new password, bound to the digest the account keeps, so
# it works once.
pub fn reset db id email kept
  time/now
    .> (t -> mail/url_for "/resets/{reset_token id kept t}/edit")
    .> (link -> mail/delivered_later db (reset_letter email link))

# The server, with a worker beside it that sends the mail the app queues.
serving = web/serve 3000 served_routes connected