// Package oltp is quarry's forge OLTP database (0001 §8): the only
// authoritative non-DAG state. v1 holds sync users — per-repo login
// credentials for push gating. Small by construction; the litestream
// replica is the recovery story when deployment warrants it.
package oltp
import (
"database/sql"
"errors"
"fmt"
"path/filepath"
"strings"
"github.com/ncruces/go-sqlite3/driver"
)
const schema = `
CREATE TABLE IF NOT EXISTS sync_user (
repo TEXT NOT NULL,
user TEXT NOT NULL,
secret TEXT NOT NULL, -- RFC §6.3 shared secret (40 hex); plaintext never stored
PRIMARY KEY (repo, user)
);
CREATE TABLE IF NOT EXISTS sync_token (
user TEXT PRIMARY KEY,
sealed TEXT NOT NULL, -- the token itself, sealed (secret.go)
created TEXT NOT NULL
);`
type DB struct {
// db is the read pool; w is the single-connection writer (see Open).
db *sql.DB
w *sql.DB
// seal keeps secret.key beside the database (secret.go); it is
// read on the first sealed write, not at open.
seal *sealer
// teamsExist short-circuits the team lookup in EffectiveCaps on a
// forge that has no teams — which is most of them. See teamFlag.
teamsExist teamFlag
// SyncProjectCode yields a repository's fossil project code, which
// the push secret's derivation needs (RFC §6.3). The composition
// root sets it; a DB without one mints nothing, which is every
// test and every CLI path with no repositories behind it.
SyncProjectCode func(repo string) (string, bool)
}
// allMigrations is every widening statement, in the order they are
// applied. Named rather than assembled at the call site so a test can
// check the same list the open runs — a guard reading a second copy
// would go stale the moment someone adds a group.
func allMigrations() []string {
var out []string
out = append(out, repoMigrations...)
out = append(out, mirrorMigrations...)
out = append(out, accountMigrations...)
out = append(out, resetMigrations...)
out = append(out, emailMigrations...)
out = append(out, sessionMigrations...)
out = append(out, webhookMigrations...)
out = append(out, gateMigrations...)
out = append(out, moderationMigrations...)
out = append(out, ciMigrations...)
out = append(out, extCIMigrations...)
out = append(out, pushDeviceMigrations...)
out = append(out, exceptionMigrations...)
out = append(out, bucketMigrations...)
out = append(out, autoMigrations...)
out = append(out, avatarMigrations...)
out = append(out, localTicketMigrations...)
return out
}
// allSchemas is every CREATE block, for the same reason.
func allSchemas() []string {
return []string{
repoSchema, repoPlaceSchema, accountSchema, sessionSchema, ciSchema, ciDeferredSchema, mirrorSchema, idempotencySchema, ticketNumberSchema,
webhookSchema, deliverySchema, subsSchema, voteSchema, pollSchema, moderationSchema,
settingsLogSchema,
stampSchema, protectSchema, checksSchema, autoSchema, gateSchema, checkinCloseSchema, categorySchema, localTicketSchema,
readSchema, capsSchema, mentionSchema, chatReadSchema, chatRefSchema, chatReplySchema,
attnSchema, pushDeviceSchema, chatBridgeSchema, exceptionSchema, bucketSchema, attachmentSchema, localAttachSchema, avatarSchema,
emailSchema, teamSchema, forgeActSchema,
}
}
// readerDSN names the database with its pragmas in the URI, because the
// driver applies those to every connection it opens; a PRAGMA run with
// Exec after opening reaches only whichever pooled connection happened
// to be free. busy_timeout comes first because the driver documents
// that order. The reader pool keeps deferred transactions: an immediate
// one would take the write lock at BEGIN, stepping around the writer
// queue and stalling it for as long as the read lasts.
func readerDSN(path string) string {
return "file:" + path + "?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)"
}
// writerDSN adds _txlock=immediate, so every write transaction takes the
// write lock at BEGIN: a deferred transaction that reads and then writes
// fails with SQLITE_BUSY_SNAPSHOT when another writer commits in
// between, and no busy timeout can rescue it.
func writerDSN(path string) string {
return readerDSN(path) + "&_txlock=immediate"
}
func Open(path string) (_ *DB, err error) {
// Writes and reads use separate handles over the same file. The
// writer holds one connection, so writers queue in Go instead of
// polling SQLite's lock, and a write never waits behind a read.
// Every Exec and Begin in this package goes to the writer; no
// transaction body touches d.db, which is what keeps one connection
// from deadlocking on itself. Reads stay on a pool because WAL lets
// them run beside the writer.
db, err := driver.Open(writerDSN(path))
if err != nil {
return nil, err
}
db.SetMaxOpenConns(1)
// Any failed return closes both handles; Close is idempotent, so the
// explicit closes on the schema paths below are harmless.
var read *sql.DB
defer func() {
if err != nil {
if read != nil {
read.Close()
}
db.Close()
}
}()
if _, err := db.Exec(schema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: schema: %w", err)
}
if _, err := db.Exec(accountSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: account schema: %w", err)
}
if _, err := db.Exec(webhookSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: webhook schema: %w", err)
}
if _, err := db.Exec(deliverySchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: webhook delivery schema: %w", err)
}
if _, err := db.Exec(stampSchema); err != nil {
return nil, fmt.Errorf("oltp: stamp schema: %w", err)
}
if _, err := db.Exec(inviteSchema); err != nil {
return nil, err
}
if _, err := db.Exec(sessionSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: session schema: %w", err)
}
if _, err := db.Exec(ciSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: ci schema: %w", err)
}
if _, err := db.Exec(mirrorSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: mirror schema: %w", err)
}
if _, err := db.Exec(forgeActSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: forge act schema: %w", err)
}
if _, err := db.Exec(repoSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: repo schema: %w", err)
}
if _, err := db.Exec(extCISchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: external ci schema: %w", err)
}
if _, err := db.Exec(ciSecretSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: ci secret schema: %w", err)
}
if _, err := db.Exec(checksSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: checks schema: %w", err)
}
if _, err := db.Exec(autoSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: auto-accept schema: %w", err)
}
if _, err := db.Exec(gateSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: gate schema: %w", err)
}
if _, err := db.Exec(checkinCloseSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: checkin close schema: %w", err)
}
if _, err := db.Exec(localTicketSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: local ticket schema: %w", err)
}
if _, err := db.Exec(companionSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: companion schema: %w", err)
}
if _, err := db.Exec(inboundSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: inbound schema: %w", err)
}
if _, err := db.Exec(protectSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: protect schema: %w", err)
}
if _, err := db.Exec(voteSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: vote schema: %w", err)
}
if _, err := db.Exec(categorySchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: category schema: %w", err)
}
if _, err := db.Exec(pollSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: poll schema: %w", err)
}
if _, err := db.Exec(settingsLogSchema); err != nil {
return nil, err
}
if _, err := db.Exec(moderationSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: moderation schema: %w", err)
}
if _, err := db.Exec(savedViewSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: saved view schema: %w", err)
}
if _, err := db.Exec(subsSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: subs schema: %w", err)
}
if _, err := db.Exec(readSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: read schema: %w", err)
}
if _, err := db.Exec(ciDeferredSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: ci deferred schema: %w", err)
}
if _, err := db.Exec(idempotencySchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: idempotency schema: %w", err)
}
if _, err := db.Exec(ticketNumberSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: ticket number schema: %w", err)
}
if _, err := db.Exec(repoPlaceSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: repo place schema: %w", err)
}
if _, err := db.Exec(teamSchema); err != nil {
return nil, err
}
if _, err := db.Exec(capsSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: caps schema: %w", err)
}
if _, err := db.Exec(mentionSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: mention schema: %w", err)
}
if _, err := db.Exec(chatRefSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: chat ref schema: %w", err)
}
if _, err := db.Exec(chatReplySchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: chat reply schema: %w", err)
}
if _, err := db.Exec(chatReadSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: chat read schema: %w", err)
}
if _, err := db.Exec(exceptionSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: exception schema: %w", err)
}
if _, err := db.Exec(attnSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: attention schema: %w", err)
}
if _, err := db.Exec(pushDeviceSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: push device schema: %w", err)
}
if _, err := db.Exec(chatBridgeSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: chat bridge schema: %w", err)
}
if _, err := db.Exec(bucketSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: bucket schema: %w", err)
}
if _, err := db.Exec(attachmentSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: attachment schema: %w", err)
}
if _, err := db.Exec(localAttachSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: local attachment schema: %w", err)
}
if _, err := db.Exec(avatarSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: avatar schema: %w", err)
}
if _, err := db.Exec(emailSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: email schema: %w", err)
}
for _, alter := range allMigrations() {
if _, err := db.Exec(alter); err != nil && !strings.Contains(err.Error(), "duplicate column") {
db.Close()
return nil, fmt.Errorf("oltp: migrate: %w", err)
}
}
if _, err := db.Exec(fossilUserSchema); err != nil {
db.Close()
return nil, fmt.Errorf("oltp: fossil user schema: %w", err)
}
read, err = driver.Open(readerDSN(path))
if err != nil {
return nil, err
}
out := &DB{db: read, w: db, seal: &sealer{path: filepath.Join(filepath.Dir(path), sealKeyFile)}}
// Legacy role rows become cap strings on every open — idempotent,
// and explicit caps are never clobbered (RFC 0004 §5).
if err := out.MigrateRolesToCaps(); err != nil {
return nil, fmt.Errorf("oltp: roles→caps: %w", err)
}
// The praise table's rows become stamps. Before any caller reads.
if err := out.carryPraise(); err != nil {
return nil, fmt.Errorf("oltp: praise→stamp: %w", err)
}
// After every schema above, and after the mirror and bridge tables
// it reads: which mirrors have chat on is settled once, here.
if err := out.mirrorChatOff(); err != nil {
return nil, fmt.Errorf("oltp: mirror chat default: %w", err)
}
if err := out.seedFleetDefault(); err != nil {
return nil, fmt.Errorf("oltp: fleet default: %w", err)
}
return out, nil
}
func (d *DB) Close() error {
return errors.Join(d.db.Close(), d.w.Close())
}
// SetSyncUser stores (replacing) one repo user's shared secret.
func (d *DB) SetSyncUser(repo, user, secret string) error {
_, err := d.w.Exec(`INSERT OR REPLACE INTO sync_user (repo, user, secret)
VALUES (?, ?, ?)`, repo, user, secret)
return err
}
// Secret returns the shared secret a push must be signed with.
//
// Two sources, in order. A sync_user row wins when one exists: those
// are hand-issued, they predate tokens, and somebody is relying on
// one. Otherwise the account's push token is unsealed and the
// repository's secret derived from it on the spot (synctoken.go).
//
// Deriving here rather than storing a row per repository is what keeps
// the credential from going stale: a repository made after the token
// was minted, or a grant made a minute ago, is answered correctly
// without anybody re-issuing anything. The letters are consulted on
// every call for the same reason — revoking write has to stop a push
// immediately, not at the next mint.
func (d *DB) Secret(repo, user string) (string, bool) {
var s string
err := d.db.QueryRow(`SELECT secret FROM sync_user WHERE repo = ? AND user = ?`,
repo, user).Scan(&s)
if err == nil {
return s, true
}
if d.SyncProjectCode == nil {
return "", false
}
if !d.EffectiveCaps(user, repo).HasAny("i") {
return "", false
}
token, ok := d.syncToken(user)
if !ok {
return "", false
}
code, ok := d.SyncProjectCode(repo)
if !ok || code == "" {
return "", false
}
return syncSecret(code, user, token), true
}
// MayWrite reports the check-in letter alone, for the half of a push
// refusal that Secret cannot distinguish: Secret answers false both for
// a person with the letter and no credential and for a person with a
// credential and no letter, and those want opposite advice.
func (d *DB) MayWrite(repo, user string) bool {
return d.EffectiveCaps(user, repo).HasAny("i")
}
// Gated reports whether a push must authenticate.
//
// The letters decide, and only the letters. An unauthenticated pusher
// is a signed-out visitor, and a signed-out visitor holds the `nobody`
// category alone — `anonymous` is unioned only for an account that
// signed in — so EffectiveCaps with no account is exactly the right
// question. It is also the expanded set, so a category that reaches i
// through u or v still opens the gate, and one that does not cannot.
//
// Stock fossil agrees: 2.28 ships nobody=gjorz and anonymous=hz, so
// pushing without an account is an opt-in an administrator performs,
// not a default anyone inherits.
//
// This used to ALSO answer true whenever the repository had any sync
// users, which reads as a sensible "somebody configured auth here". It
// was not. It let the presence of a credential row become policy, and
// the two clauses were answering different questions — "must a push
// authenticate" and "has anybody been issued a credential" — so the
// gate closed on repositories whose letters said otherwise and stayed
// closed, because nothing but a hand-run CLI could issue the row it
// was now demanding. Whether a credential exists is MintSyncSecrets'
// business; whether one is required is this function's, and they are
// not the same question.
//
// Fails closed without a special case: a database that cannot answer
// yields empty caps, empty caps hold no i, and no i is gated.
func (d *DB) Gated(repo string) bool {
return !d.EffectiveCaps("", repo).HasAny("i")
}