package oltp
import (
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"time"
)
// Local tickets (docs/local-tickets.md): a staging tier in front of the
// fossil. Fossil tickets are artifacts and artifacts are forever, which
// is right for provenance and wrong for an open intake door. A local
// ticket lives here in oltp — same fields, same views — and enters the
// zit store only when an admin promotes it. Until then it can simply be
// deleted, which no artifact ever can.
//
// Storage is fossil-shaped on purpose: a ticket is nothing but its
// ordered changes, and the current state is a fold over them. That
// keeps the write path an append, mirrors the semantics readers already
// know from the artifact tier, and costs nothing at intake-queue scale.
const localTicketSchema = `
CREATE TABLE IF NOT EXISTS local_ticket (
repo TEXT NOT NULL,
id TEXT NOT NULL, -- minted 40-hex; same address shape as fossil ids
created TEXT NOT NULL,
updated TEXT NOT NULL,
author TEXT NOT NULL DEFAULT '',
addr TEXT NOT NULL DEFAULT '', -- submitter address, for the intake rate limit
promoted_to TEXT NOT NULL DEFAULT '', -- fossil ticket id once promoted
PRIMARY KEY (repo, id)
);
CREATE TABLE IF NOT EXISTS local_ticket_change (
repo TEXT NOT NULL,
id TEXT NOT NULL,
seq INTEGER NOT NULL,
date TEXT NOT NULL,
user TEXT NOT NULL,
fields TEXT NOT NULL, -- JSON object, field name -> value
PRIMARY KEY (repo, id, seq)
);`
// LocalTicketChange is one recorded change, fields as written.
type LocalTicketChange struct {
Date string `json:"date"`
User string `json:"user"`
Fields map[string]string `json:"fields"`
}
// LocalComment is one discussion entry, lifted from a change's icomment
// the way the index lifts fossil's.
type LocalComment struct {
Date string `json:"date"`
Login string `json:"login"`
Mimetype string `json:"mimetype"`
Body string `json:"body"`
}
// LocalTicket is the folded state plus history — the local tier's
// answer to index.TicketState.
type LocalTicket struct {
ID string `json:"id"`
Created string `json:"created"`
Updated string `json:"updated"`
Author string `json:"author"`
PromotedTo string `json:"promotedTo,omitempty"`
Fields map[string]string `json:"fields"`
Comments []LocalComment `json:"comments,omitempty"`
Changes []LocalTicketChange `json:"changes,omitempty"`
}
// localPerChange mirrors the index's perChange set: fields that
// describe one change rather than the ticket, kept out of the fold.
var localPerChange = map[string]bool{
"icomment": true, "mimetype": true, "login": true, "username": true,
}
// localNow matches the index's date shape so merged listings sort by
// plain string comparison.
func localNow() string { return time.Now().UTC().Format("2006-01-02T15:04:05") }
// LocalTicketNew files a local ticket and returns its minted id.
func (d *DB) LocalTicketNew(repo, user, addr string, fields map[string]string) (string, error) {
raw := make([]byte, 20)
if _, err := rand.Read(raw); err != nil {
return "", err
}
id := hex.EncodeToString(raw)
now := localNow()
enc, err := json.Marshal(fields)
if err != nil {
return "", err
}
if _, err := d.w.Exec(`INSERT INTO local_ticket (repo, id, created, updated, author, addr)
VALUES (?, ?, ?, ?, ?, ?)`, repo, id, now, now, user, addr); err != nil {
return "", err
}
_, err = d.w.Exec(`INSERT INTO local_ticket_change (repo, id, seq, date, user, fields)
VALUES (?, ?, 1, ?, ?, ?)`, repo, id, now, user, string(enc))
return id, err
}
// LocalTicketSet appends a change to a local ticket. The id is the full
// stored id; resolve prefixes with LocalTicketByPrefix first.
func (d *DB) LocalTicketSet(repo, id, user string, fields map[string]string) error {
enc, err := json.Marshal(fields)
if err != nil {
return err
}
now := localNow()
res, err := d.w.Exec(`INSERT INTO local_ticket_change (repo, id, seq, date, user, fields)
SELECT repo, id, (SELECT COALESCE(MAX(seq),0)+1 FROM local_ticket_change WHERE repo = ? AND id = ?), ?, ?, ?
FROM local_ticket WHERE repo = ? AND id = ? AND promoted_to = ''`,
repo, id, now, user, string(enc), repo, id)
if err != nil {
return err
}
if n, _ := res.RowsAffected(); n == 0 {
return fmt.Errorf("no local ticket %.12s", id)
}
_, err = d.w.Exec(`UPDATE local_ticket SET updated = ? WHERE repo = ? AND id = ?`, now, repo, id)
return err
}
// LocalTicketByPrefix folds one local ticket. A prefix matching more
// than one id matches none, like every other prefix resolver here.
func (d *DB) LocalTicketByPrefix(repo, prefix string) (LocalTicket, bool, error) {
ts, err := d.localTickets(repo, `AND t.id LIKE ? ESCAPE '\'`, likePrefix(prefix))
if err != nil || len(ts) != 1 {
return LocalTicket{}, false, err
}
return ts[0], true, nil
}
// LocalTickets folds a repository's unpromoted local tickets, newest
// updated first.
func (d *DB) LocalTickets(repo string) ([]LocalTicket, error) {
return d.localTickets(repo, `AND t.promoted_to = ''`)
}
func (d *DB) localTickets(repo, where string, args ...any) ([]LocalTicket, error) {
rows, err := d.db.Query(`SELECT t.id, t.created, t.updated, t.author, t.promoted_to,
c.date, c.user, c.fields
FROM local_ticket t JOIN local_ticket_change c ON c.repo = t.repo AND c.id = t.id
WHERE t.repo = ? `+where+` ORDER BY t.updated DESC, t.id, c.seq`,
append([]any{repo}, args...)...)
if err != nil {
return nil, err
}
defer rows.Close()
var out []LocalTicket
var cur *LocalTicket
for rows.Next() {
var id, created, updated, author, promoted, date, user, enc string
if err := rows.Scan(&id, &created, &updated, &author, &promoted, &date, &user, &enc); err != nil {
return nil, err
}
if cur == nil || cur.ID != id {
out = append(out, LocalTicket{ID: id, Created: created, Updated: updated,
Author: author, PromotedTo: promoted, Fields: map[string]string{}})
cur = &out[len(out)-1]
}
fields := map[string]string{}
if err := json.Unmarshal([]byte(enc), &fields); err != nil {
return nil, err
}
cur.Changes = append(cur.Changes, LocalTicketChange{Date: date, User: user, Fields: fields})
for k, v := range fields {
if !localPerChange[k] {
cur.Fields[k] = v
}
}
if body := fields["icomment"]; body != "" {
cur.Comments = append(cur.Comments, LocalComment{
Date: date, Login: user, Mimetype: fields["mimetype"], Body: body})
}
}
return out, rows.Err()
}
// LocalTicketDiscard deletes a local ticket outright — the ability the
// artifact tier can never offer.
func (d *DB) LocalTicketDiscard(repo, id string) error {
if _, err := d.w.Exec(`DELETE FROM local_ticket_change WHERE repo = ? AND id = ?`, repo, id); err != nil {
return err
}
_, err := d.w.Exec(`DELETE FROM local_ticket WHERE repo = ? AND id = ?`, repo, id)
return err
}
// LocalTicketPromoted tombstones a promoted ticket: the changes go, the
// row stays as the redirect from the old id to the fossil one.
func (d *DB) LocalTicketPromoted(repo, id, fossilID string) error {
if _, err := d.w.Exec(`UPDATE local_ticket SET promoted_to = ?, updated = ?
WHERE repo = ? AND id = ?`, fossilID, localNow(), repo, id); err != nil {
return err
}
_, err := d.w.Exec(`DELETE FROM local_ticket_change WHERE repo = ? AND id = ?`, repo, id)
return err
}
// localTicketMigrations widen local_ticket for stores that predate a
// column.
var localTicketMigrations = []string{
// When a promotion claimed the ticket; '' when none holds it.
`ALTER TABLE local_ticket ADD COLUMN claimed TEXT NOT NULL DEFAULT ''`,
}
// claimLease is how long a promotion's claim holds. A promotion that
// died between claiming and tombstoning (a crash, a killed process)
// never releases its claim, and without a lease the ticket could never
// be promoted again.
const claimLease = 10 * time.Minute
// LocalTicketClaim reserves an unpromoted ticket for one promotion.
// The conditional UPDATE is the decision: of two promotions racing on
// one ticket exactly one changes the row, so exactly one goes on to
// write a ticket artifact. false means another promotion holds it or
// it is already promoted.
func (d *DB) LocalTicketClaim(repo, id string) (bool, error) {
now := time.Now().UTC()
res, err := d.w.Exec(`UPDATE local_ticket SET claimed = ?
WHERE repo = ? AND id = ? AND promoted_to = '' AND (claimed = '' OR claimed < ?)`,
now.Format("2006-01-02T15:04:05"), repo, id, now.Add(-claimLease).Format("2006-01-02T15:04:05"))
if err != nil {
return false, err
}
n, err := res.RowsAffected()
return n == 1, err
}
// LocalTicketRelease gives a claim back after a promotion that failed
// before it wrote anything, so the ticket can be promoted again now
// rather than when the lease runs out.
func (d *DB) LocalTicketRelease(repo, id string) error {
_, err := d.w.Exec(`UPDATE local_ticket SET claimed = '' WHERE repo = ? AND id = ? AND promoted_to = ''`, repo, id)
return err
}
// LocalTicketRedirect answers where a promoted local id went.
func (d *DB) LocalTicketRedirect(repo, prefix string) (string, bool) {
var to string
err := d.db.QueryRow(`SELECT promoted_to FROM local_ticket
WHERE repo = ? AND id LIKE ? ESCAPE '\' AND promoted_to != ''`,
repo, likePrefix(prefix)).Scan(&to)
return to, err == nil && to != ""
}
// LocalIntakeCount counts what one address filed since a moment — the
// number the intake rate limit compares.
func (d *DB) LocalIntakeCount(repo, addr, since string) (int, error) {
var n int
err := d.db.QueryRow(`SELECT COUNT(*) FROM local_ticket
WHERE repo = ? AND addr = ? AND created >= ?`, repo, addr, since).Scan(&n)
return n, err
}
// likePrefix escapes LIKE wildcards in a caller-supplied prefix.
func likePrefix(p string) string {
out := make([]byte, 0, len(p)+2)
for i := 0; i < len(p); i++ {
if p[i] == '%' || p[i] == '_' || p[i] == '\\' {
out = append(out, '\\')
}
out = append(out, p[i])
}
return string(append(out, '%'))
}