#moontask

#Declaring a task

#task.def(name="orders.place", queue="orders", retries="1")
pub fn place(order : Order, ctx : @moontask.Context) -> String raise {
  ...
}

let tasks = register_tasks(app)
let receipt = tasks.place.send({ id: "a1", cups: 2 }).get(timeout=5000L)

rule(name: "tasks", command: "moonx moonbitstack/moontask/cmd/tasks -C \"$mod_dir\" \"$input\" -o \"$output\"")

dev_build(rule: "tasks", input: "jobs.mbt", output: "jobs.g.mbt")
formatter(ignore: [ "jobs.g.mbt" ])

#Without the annotation

async fn first_run() -> Unit {
  let now = fn() { @env.now().reinterpret_as_int64() }
  let broker = @moontask.MemoryBroker::new(clock=now)
  let app = @moontask.App::new(broker, backend=broker, counter=broker, clock=now)

  let greet = app.task("greet", fn(name : String, ctx : @moontask.Context) {
    "hello \{name}, attempt \{ctx.attempt}"
  })

  let worker = app.worker(concurrency=4)
  @async.with_task_group(async fn(group) {
    group.spawn_bg(async fn() { worker.run() }, no_wait=true)
    let handle = greet.send("ada")
    println(handle.get(timeout=5000L))
    worker.stop()
  })
}

#What is in the module

#A task's policy

#The canvas

async fn canvas() -> Unit {
  let now = fn() { 1_700_000_000_000L }
  let broker = @moontask.MemoryBroker::new(clock=now)
  let app = @moontask.App::new(broker, backend=broker, counter=broker, clock=now)
  let step = app.task("step", fn(value : Int32, _) { value * 10 })
  let total = app.task("total", fn(values : Array[Int32], _) {
    let mut sum = 0
    for value in values {
      sum = sum + value
    }
    sum
  })
  let chained = @moontask.chain([step.sig(1), step.sig(2), step.sig(3)]).send(app)
  let grouped = @moontask.group([step.sig(4), step.sig(5)]).send(app)
  let joined = @moontask.chord(
    [step.sig(1), step.sig(2), step.sig(3)],
    total.sig([]),
  ).send(app)
  let mapped = step.map([6, 7]).send(app)
  let starred = total.starmap([[1, 2], [3, 4]]).send(app)
  let chunked = step.chunks([[1], [2], [3]], size=2).send(app)
  println(
    "\{chained.ids().length()} \{grouped.ids().length()} \{joined.ids().length()}",
  )
  println(
    "\{mapped.ids().length()} \{starred.ids().length()} \{chunked.ids().length()}",
  )
}

#On a rhythm

fn rhythm() -> Unit {
  let now = fn() { 1_700_000_000_000L }
  let broker = @moontask.MemoryBroker::new(clock=now)
  let app = @moontask.App::new(broker, backend=broker, counter=broker, clock=now)
  let tick = app.task("tick", fn(_ : Unit, _) { () })
  let nightly = @cron.parse("0 3 * * *") catch { _ => abort("valid expression") }
  let beat = app.beat()
  beat.add("tick", tick, (), @moontask.Schedule::Every(interval=1000L))
  beat.add("nightly", tick, (), @moontask.Schedule::Cron(expr=nightly))
  println(beat.pending().length())
}

#Workflows

async fn workflow() -> Unit {
  let now = fn() { 1_700_000_000_000L }
  let broker = @moontask.MemoryBroker::new(clock=now)
  let app = @moontask.App::new(broker, backend=broker, counter=broker, clock=now)
  let extract = app.task("extract", fn(_ : Unit, _) -> Json { { "rows": 3 } })
  let load = app.task("load", fn(args : Json, _) -> Json {
    { "written": 3, "from": args }
  })
  let flow = @flow.Flow::new(
    "etl",
    schedule=@moontask.Schedule::Every(interval=3_600_000L),
  )
  flow.step("extract", extract)
  flow.step("load", load, deps=["extract"])
  let worker = app.worker()
  @async.with_task_group(async fn(group) {
    group.spawn_bg(async fn() { worker.run() }, no_wait=true)
    let run = flow.run(app, 1_700_000_000_000L)
    println(run.to_string())
    println(run.output_of("load").unwrap().stringify())
    worker.stop()
  })
}

#Results and run records on SQL

import {
  "moonbitstack/moontask",
  "moonbitstack/moontask/flow",
  "moonbitstack/moontask-sql/sql",
  "moonbitstack/moonpostgres",
}

async fn sql_backend {
  let conn = @moonpostgres.PgConn::connect("127.0.0.1", 5432, "postgres", "postgres", "test")
  let store = @sql.Store::new(conn, clock=fn() { @env.now().reinterpret_as_int64() })
  store.ensure()
  let app = @moontask.App::new(broker, backend=store, clock=fn() { @env.now().reinterpret_as_int64() })
  let flow = @flow.Flow::new("etl")
  flow.record(store)
}

let db = @sqlite.SqliteDriver::open("queue.db")
let store = @sql.Store::inline(db, clock=fn() { @env.now().reinterpret_as_int64() })
store.ensure()

#Across processes

let conn = @socket.Connection::connect(@moonredis.ClientOptions::at("127.0.0.1", 6379))
let broker = @redis.Broker::new(conn, clock=now, consumer="worker-7")
let app = @moontask.App::new(broker, backend=broker.store(), counter=broker.store(), clock=now)

let broker = @redis.Broker::celery(conn, clock=now, consumer="worker-7")

#Writing a broker

pub(open) trait Broker {
  async fn publish(Self, queue : String, message : @spec.Message) -> Unit raise @spec.Refused
  async fn consume(Self, queues : ArrayView[String], prefetch~ : Int32) -> @spec.Delivery raise @spec.Refused
  async fn ack(Self, delivery : @spec.Delivery) -> Unit raise @spec.Refused
  async fn reject(Self, delivery : @spec.Delivery, requeue~ : Bool) -> Unit raise @spec.Refused
}

#Examples

moon run src/examples/01-queue
moon run src/examples/02-canvas
moon run src/examples/03-retry
moon run src/examples/04-flow
moon run src/examples/05-floor
moon run src/examples/06-annotated
moon run src/examples/07-fanout

#License