Artifact [d5cb356acd]

Artifact d5cb356acdf61020b56e468044bb6f725f06e0998cf50fcd83be5468ebb7e8cb:

• File internal/oltp/oltp.go — part of check-in [550b7af81c] at 2026-08-16 19:38:20 on branch trunk — webhooks: the standard outbound surface (0001 Principle 6)

  Activity events reach third parties over signed HTTP, so an integrator never has to speak NATS. Subscriptions live in forge OLTP; the signing secret is printed once at creation and never readable again.

  Delivery is durable rather than a fire-and-forget goroutine: each (event, subscription) is one JetStream message, acked on 2xx, NAKed with growing backoff, MaxDeliver as the dead-letter boundary. An endpoint down during a deploy still gets its events — proven live by writing with the receiver stopped and seeing the delivery land when it returned.

  0003 §3 names dagnats for this class of work; it runs on the JetStream quarry already embeds, so this uses that directly and leaves dagnats as the upgrade path when v3 brings the engine in for CI.

  Mentions are not forwarded: they address a person, not a repository. (user: dmestas, size: 2219) [annotate] [blame] [check-ins using]

// 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"
	"fmt"

	"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)
);`

type DB struct {
	db *sql.DB
}

func Open(path string) (*DB, error) {
	db, err := driver.Open("file:" + path)
	if err != nil {
		return nil, err
	}
	if _, err := db.Exec(`PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000;`); err != nil {
		db.Close()
		return nil, fmt.Errorf("oltp: pragmas: %w", err)
	}
	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)
	}
	return &DB{db: db}, nil
}

func (d *DB) Close() error { return d.db.Close() }

// SetSyncUser stores (replacing) one repo user's shared secret.
func (d *DB) SetSyncUser(repo, user, secret string) error {
	_, err := d.db.Exec(`INSERT OR REPLACE INTO sync_user (repo, user, secret)
		VALUES (?, ?, ?)`, repo, user, secret)
	return err
}

// Secret returns the stored shared secret for one repo user.
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)
	return s, err == nil
}

// Gated reports whether a repo has any sync users — the v1 policy
// switch: no users means anonymous push stays allowed (local use),
// one or more users means pushes must authenticate.
func (d *DB) Gated(repo string) bool {
	var n int
	if err := d.db.QueryRow(`SELECT count(*) FROM sync_user WHERE repo = ?`, repo).Scan(&n); err != nil {
		return false
	}
	return n > 0
}