Skip to content
quarry Artifact [26f48ccbb5]
Login

Artifact [26f48ccbb5]

Artifact 26f48ccbb5664c659623e6577de576447e4950435c7a063d49541c66de278e12:


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.db.Exec(`INSERT INTO local_ticket (repo, id, created, updated, author, addr)
		VALUES (?, ?, ?, ?, ?, ?)`, repo, id, now, now, user, addr); err != nil {
		return "", err
	}
	_, err = d.db.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.db.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.db.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.db.Exec(`DELETE FROM local_ticket_change WHERE repo = ? AND id = ?`, repo, id); err != nil {
		return err
	}
	_, err := d.db.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.db.Exec(`UPDATE local_ticket SET promoted_to = ?, updated = ?
		WHERE repo = ? AND id = ?`, fossilID, localNow(), repo, id); err != nil {
		return err
	}
	_, err := d.db.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.db.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.db.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, '%'))
}