//// Durable admissions. Pending inputs and their receipts share a transaction;
//// receipts deliberately have no session foreign key and survive deletion.
import albedo/daemon/store
import albedo/daemon/usage
import gleam/dynamic/decode
import gleam/list
import gleam/option.{type Option, None, Some}
import gleam/result
import sqlight
pub const retention_ms = 604_800_000
pub const schema = "CREATE TABLE IF NOT EXISTS operations(id TEXT PRIMARY KEY,fingerprint TEXT NOT NULL,kind TEXT NOT NULL,target TEXT NOT NULL,status TEXT NOT NULL CHECK(status IN ('accepted','rejected')),http_status INTEGER NOT NULL,response TEXT NOT NULL,created_at INTEGER NOT NULL,terminal_at INTEGER,delivery TEXT CHECK(delivery IN ('pending','committed','cancelled')),blocking_reason TEXT,committed_seq INTEGER); CREATE INDEX IF NOT EXISTS operations_expiry ON operations(terminal_at) WHERE terminal_at IS NOT NULL; CREATE TABLE IF NOT EXISTS pending_inputs(seq INTEGER PRIMARY KEY AUTOINCREMENT,operation_id TEXT NOT NULL UNIQUE REFERENCES operations(id),session TEXT NOT NULL REFERENCES sessions(id),payload BLOB NOT NULL,accepted_at INTEGER NOT NULL); CREATE INDEX IF NOT EXISTS pending_inputs_session ON pending_inputs(session,seq); CREATE TABLE IF NOT EXISTS submission_events(session TEXT NOT NULL REFERENCES sessions(id),seq INTEGER NOT NULL,operation_id TEXT NOT NULL UNIQUE,payload BLOB NOT NULL); CREATE INDEX IF NOT EXISTS submission_events_session ON submission_events(session,seq); CREATE TABLE IF NOT EXISTS continuation_markers(operation_id TEXT PRIMARY KEY,session TEXT NOT NULL REFERENCES sessions(id),seq INTEGER NOT NULL,payload BLOB NOT NULL); CREATE INDEX IF NOT EXISTS continuation_markers_session ON continuation_markers(session,seq);"
pub type Request {
Request(id: String, fingerprint: String, kind: String, target: String)
}
pub type Receipt {
Receipt(
operation: Request,
status: String,
http_status: Int,
response: String,
created_at: Int,
delivery: Option(String),
blocking_reason: Option(String),
committed_seq: Option(Int),
)
}
pub type Pending {
Pending(id: String, session: String, payload: BitArray, accepted_at: Int)
}
/// Original upload metadata; the transcript and image store own its content.
pub type ImageMetadata {
ImageMetadata(mime_type: String, width: Int, height: Int, bytes: Int)
}
pub type Display {
Display(
text: String,
source: String,
client_id: String,
operation_id: Option(String),
image: Option(ImageMetadata),
)
}
/// The transcript offset is absent for continuations, which append no row.
pub type Commit {
Commit(id: String, offset: Option(Int), display: Display)
}
@external(erlang, "erlang", "term_to_binary")
fn encode_display(display: Display) -> BitArray
@external(erlang, "albedo_operations", "decode_display")
fn decode_display(payload: BitArray) -> Display
fn receipt_decoder() -> decode.Decoder(Receipt) {
use id <- decode.field(0, decode.string)
use fingerprint <- decode.field(1, decode.string)
use kind <- decode.field(2, decode.string)
use target <- decode.field(3, decode.string)
use status <- decode.field(4, decode.string)
use http_status <- decode.field(5, decode.int)
use response <- decode.field(6, decode.string)
use created_at <- decode.field(7, decode.int)
use delivery <- decode.field(8, decode.optional(decode.string))
use blocking <- decode.field(9, decode.optional(decode.string))
use committed_seq <- decode.field(10, decode.optional(decode.int))
decode.success(Receipt(
Request(id, fingerprint, kind, target),
status,
http_status,
response,
created_at,
delivery,
blocking,
committed_seq,
))
}
pub fn lookup(
ledger: store.Store,
id: String,
) -> Result(Option(Receipt), String) {
store.query(ledger, lookup_in(_, id))
}
pub fn lookup_in(
db: sqlight.Connection,
id: String,
) -> Result(Option(Receipt), String) {
use rows <- result.try(store.rows(
db,
"SELECT id,fingerprint,kind,target,status,http_status,response,created_at,delivery,blocking_reason,committed_seq FROM operations WHERE id=?",
[sqlight.text(id)],
receipt_decoder(),
))
Ok(list.first(rows) |> option.from_result)
}
/// Existing receipts win over all mutable validation and UUID age checks.
pub fn check(
ledger: store.Store,
request: Request,
) -> Result(Option(Receipt), String) {
store.query(ledger, check_in(_, request))
}
pub fn check_in(
db: sqlight.Connection,
request: Request,
) -> Result(Option(Receipt), String) {
use receipt <- result.try(lookup_in(db, request.id))
case receipt {
Some(receipt) -> {
// Creation learns its target when the session row is allocated. Its
// fingerprint identifies the original request before that allocation.
let original = receipt.operation
let matches =
original.id == request.id
&& original.fingerprint == request.fingerprint
&& original.kind == request.kind
&& { request.kind == "create" || original.target == request.target }
case matches {
True -> Ok(Some(receipt))
False -> Error("operation_conflict")
}
}
None -> validate_id(request.id, usage.now()) |> result.replace(None)
}
}
/// Runs the mutation and saves its admission response on this same connection.
/// A duplicate does not call the mutation, even when the session was deleted.
pub fn admit(
ledger: store.Store,
request: Request,
http_status: Int,
response: String,
delivery: Option(String),
mutate: fn(sqlight.Connection) -> Result(Nil, String),
) -> Result(Receipt, String) {
store.query(ledger, fn(db) {
store.transaction(db, fn() {
use previous <- result.try(check_in(db, request))
case previous {
Some(receipt) -> Ok(receipt)
None -> {
let now = usage.now()
use _ <- result.try(insert(
db,
request,
"accepted",
http_status,
response,
now,
delivery,
))
use _ <- result.try(mutate(db))
Ok(Receipt(
request,
"accepted",
http_status,
response,
now,
delivery,
None,
None,
))
}
}
})
})
}
pub fn admit_pending(
ledger: store.Store,
request: Request,
payload: BitArray,
http_status: Int,
response: String,
) -> Result(Receipt, String) {
admit(ledger, request, http_status, response, Some("pending"), fn(db) {
store.run(
db,
"INSERT INTO pending_inputs(operation_id,session,payload,accepted_at) VALUES(?,?,?,?)",
[
sqlight.text(request.id),
sqlight.text(request.target),
sqlight.blob(payload),
sqlight.int(usage.now()),
],
)
})
}
pub fn reject(
ledger: store.Store,
request: Request,
http_status: Int,
response: String,
) -> Result(Receipt, String) {
store.query(ledger, fn(db) {
store.transaction(db, fn() {
use previous <- result.try(check_in(db, request))
case previous {
Some(receipt) -> Ok(receipt)
None -> {
let now = usage.now()
use _ <- result.try(insert(
db,
request,
"rejected",
http_status,
response,
now,
None,
))
Ok(Receipt(
request,
"rejected",
http_status,
response,
now,
None,
None,
None,
))
}
}
})
})
}
fn insert(
db: sqlight.Connection,
request: Request,
status: String,
http_status: Int,
response: String,
now: Int,
delivery: Option(String),
) -> Result(Nil, String) {
store.run(
db,
"INSERT INTO operations(id,fingerprint,kind,target,status,http_status,response,created_at,terminal_at,delivery) VALUES(?,?,?,?,?,?,?,?,?,?)",
[
sqlight.text(request.id),
sqlight.text(request.fingerprint),
sqlight.text(request.kind),
sqlight.text(request.target),
sqlight.text(status),
sqlight.int(http_status),
sqlight.text(response),
sqlight.int(now),
sqlight.nullable(sqlight.int, case delivery {
Some("pending") -> None
_ -> Some(now)
}),
sqlight.nullable(sqlight.text, delivery),
],
)
}
pub fn pending(
ledger: store.Store,
session: String,
) -> Result(List(Pending), String) {
store.read(
ledger,
"SELECT operation_id,session,payload,accepted_at FROM pending_inputs WHERE session=? ORDER BY seq",
[sqlight.text(session)],
{
use id <- decode.field(0, decode.string)
use session <- decode.field(1, decode.string)
use payload <- decode.field(2, decode.bit_array)
use accepted_at <- decode.field(3, decode.int)
decode.success(Pending(id, session, payload, accepted_at))
},
)
}
pub fn pending_sessions(ledger: store.Store) -> Result(List(String), String) {
store.read(
ledger,
"SELECT session FROM pending_inputs GROUP BY session ORDER BY MIN(seq)",
[],
decode.field(0, decode.string, decode.success),
)
}
pub fn block(
ledger: store.Store,
ids: List(String),
reason: String,
) -> Result(Nil, String) {
store.query(ledger, fn(db) {
list.try_each(ids, fn(id) {
store.run(
db,
"UPDATE operations SET blocking_reason=? WHERE id=? AND delivery='pending'",
[sqlight.text(reason), sqlight.text(id)],
)
})
})
}
/// Called only within the transaction that appends these inputs.
pub fn committed(
db: sqlight.Connection,
session: String,
submissions: List(Commit),
first_seq: Int,
last_seq: Int,
now: Int,
) -> Result(Nil, String) {
list.try_each(submissions, fn(submission) {
let Commit(id, offset, display) = submission
use _ <- result.try(store.one(
db,
"SELECT operation_id FROM pending_inputs WHERE operation_id=? AND session=?",
[sqlight.text(id), sqlight.text(session)],
decode.field(0, decode.string, decode.success),
"pending operation missing",
))
let seq = case offset {
Some(offset) -> first_seq + offset
None -> last_seq
}
use _ <- result.try(case offset {
Some(_) ->
store.run(
db,
"INSERT INTO submission_events(session,seq,operation_id,payload) VALUES(?,?,?,?)",
[
sqlight.text(session),
sqlight.int(seq),
sqlight.text(id),
sqlight.blob(encode_display(display)),
],
)
None ->
store.run(
db,
"INSERT INTO continuation_markers(operation_id,session,seq,payload) VALUES(?,?,?,?)",
[
sqlight.text(id),
sqlight.text(session),
sqlight.int(seq),
sqlight.blob(encode_display(display)),
],
)
})
use _ <- result.try(
store.run(
db,
"UPDATE operations SET delivery='committed',terminal_at=?,committed_seq=?,blocking_reason=NULL WHERE id=? AND delivery='pending'",
[sqlight.int(now), sqlight.int(seq), sqlight.text(id)],
),
)
store.run(
db,
"DELETE FROM pending_inputs WHERE operation_id=? AND session=?",
[sqlight.text(id), sqlight.text(session)],
)
})
}
/// One row's display metadata without scanning the session's submissions.
pub fn committed_input(
ledger: store.Store,
session: String,
seq: Int,
) -> Result(Option(Display), String) {
use rows <- result.try(store.read(
ledger,
"SELECT payload FROM submission_events WHERE session=? AND seq=?",
[sqlight.text(session), sqlight.int(seq)],
decode.field(0, decode.bit_array, decode.success),
))
Ok(list.first(rows) |> option.from_result |> option.map(decode_display))
}
pub fn cancel(ledger: store.Store, session: String) -> Result(Nil, String) {
store.query(ledger, fn(db) {
store.transaction(db, fn() { cancel_in(db, session) })
})
}
/// Cancels only the waiting inputs selected by submission ownership.
pub fn cancel_inputs(
ledger: store.Store,
session: String,
ids: List(String),
) -> Result(Nil, String) {
store.query(ledger, fn(db) {
store.transaction(db, fn() {
list.try_each(ids, fn(id) {
use _ <- result.try(
store.run(
db,
"UPDATE operations SET delivery='cancelled',terminal_at=?,blocking_reason=NULL WHERE id IN (SELECT operation_id FROM pending_inputs WHERE session=? AND operation_id=?)",
[sqlight.int(usage.now()), sqlight.text(session), sqlight.text(id)],
),
)
store.run(
db,
"DELETE FROM pending_inputs WHERE session=? AND operation_id=?",
[sqlight.text(session), sqlight.text(id)],
)
})
})
})
}
pub fn cancel_in(
db: sqlight.Connection,
session: String,
) -> Result(Nil, String) {
use _ <- result.try(
store.run(
db,
"UPDATE operations SET delivery='cancelled',terminal_at=?,blocking_reason=NULL WHERE id IN (SELECT operation_id FROM pending_inputs WHERE session=?)",
[sqlight.int(usage.now()), sqlight.text(session)],
),
)
store.run(db, "DELETE FROM pending_inputs WHERE session=?", [
sqlight.text(session),
])
}
/// Each call removes at most 128 terminal receipts; pending work never ages out.
pub fn prune(ledger: store.Store) -> Result(Nil, String) {
store.write(
ledger,
"DELETE FROM operations WHERE id IN (SELECT id FROM operations WHERE terminal_at IS NOT NULL AND terminal_at<? ORDER BY terminal_at LIMIT 128)",
[sqlight.int(usage.now() - retention_ms)],
)
}
@external(erlang, "albedo_operations", "validate_id")
pub fn validate_id(id: String, now: Int) -> Result(Nil, String)