Artifact [08c211e5f2]

Artifact 08c211e5f2bd5303748a3cb9740506fdb4db2ad6b409f719e10454fa73d3de9d:

• File cmd/quarry/main.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: 26173) [annotate] [blame] [check-ins using]

// quarry is a multi-repository forge for zit. See docs/rfcs/.
package main

import (
	"context"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"craftdesign.group/quarry/internal/api"
	"craftdesign.group/quarry/internal/chat"
	"craftdesign.group/quarry/internal/events"
	"craftdesign.group/quarry/internal/index"
	"craftdesign.group/quarry/internal/ingest"
	"craftdesign.group/quarry/internal/lease"
	"craftdesign.group/quarry/internal/notify"
	"craftdesign.group/quarry/internal/oltp"
	"craftdesign.group/quarry/internal/ops"
	"craftdesign.group/quarry/internal/web"
	"craftdesign.group/quarry/internal/webhook"
	"craftdesign.group/quarry/internal/xfer"
	"craftdesign.group/quarry/internal/zitcli"
)

// version is stamped at build time; every page footer and the API report it
// (0001 Principle 1).
var version = "0.0.0-dev"

func main() {
	if len(os.Args) > 1 {
		switch os.Args[1] {
		case "version":
			fmt.Println("quarry", version)
			return
		case "spec":
			if err := printSpec(os.Stdout); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		case "serve":
			if err := serveCmd(os.Args[2:]); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		case "reindex":
			if err := reindexCmd(os.Args[2:]); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		case "docs":
			r := ops.NewRegistry()
			if err := api.RegisterAll(r, api.Deps{Version: version}); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			os.Stdout.Write(ops.Docs(r, "quarry"))
			return
		case "user":
			if err := userCmd(os.Args[2:]); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		case "webhook":
			if err := webhookCmd(os.Args[2:]); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		case "account":
			if err := accountCmd(os.Args[2:]); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		}
	}
	fmt.Fprintln(os.Stderr, "usage: quarry version|spec|serve|reindex|user|account|webhook|docs")
	os.Exit(2)
}

// userCmd manages sync users (0001 §7 v1 auth): `quarry user add`
// derives the RFC §6.3 shared secret from the store's project code and
// stores it in forge OLTP — plaintext is never persisted. A repo with
// at least one sync user requires authenticated pushes.
func userCmd(args []string) error {
	if len(args) < 1 || args[0] != "add" {
		return fmt.Errorf("usage: quarry user add --repo name=path --user U [--password P] [--data DIR]")
	}
	fs := flag.NewFlagSet("user add", flag.ContinueOnError)
	repos := repoFlags{}
	fs.Var(repos, "repo", "repository as name=path")
	user := fs.String("user", "", "sync login name")
	password := fs.String("password", "", "password (omit to read from QUARRY_PASSWORD)")
	data := fs.String("data", "", "derived-data directory (default: .quarry beside the repo)")
	if err := fs.Parse(args[1:]); err != nil {
		return err
	}
	if len(repos) != 1 || *user == "" {
		return fmt.Errorf("user add wants exactly one --repo and a --user")
	}
	pw := *password
	if pw == "" {
		pw = os.Getenv("QUARRY_PASSWORD")
	}
	if pw == "" {
		return fmt.Errorf("no password: use --password or QUARRY_PASSWORD")
	}
	var name, path string
	for n, p := range repos {
		name, path = n, p
	}
	dir := *data
	if dir == "" {
		dir = filepath.Join(path, ".quarry")
	}
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return err
	}
	code, err := xfer.ProjectCode(path)
	if err != nil {
		return fmt.Errorf("project code of %s: %w", name, err)
	}
	db, err := oltp.Open(filepath.Join(dir, "oltp.db"))
	if err != nil {
		return err
	}
	defer db.Close()
	if err := db.SetSyncUser(name, *user, xfer.SharedSecret(code, *user, pw)); err != nil {
		return err
	}
	fmt.Printf("sync user %q added to %s; pushes now require authentication\n", *user, name)
	return nil
}

// reindexCmd is 0001 §5's full rebuild as a verb: for each --repo, drop
// the derived tables and cursor, then replay the store. Run it with the
// same --repo/--data flags as serve, while serve is stopped (the
// embedded NATS store directory admits one process).
func reindexCmd(args []string) error {
	fs := flag.NewFlagSet("reindex", flag.ContinueOnError)
	repos := repoFlags{}
	fs.Var(repos, "repo", "repository as name=path (repeatable; bare path uses its basename)")
	data := fs.String("data", "", "derived-data directory (default: .quarry beside the first repo)")
	natsURL := fs.String("nats", "", "external NATS URL (multi-node); reindex then publishes a cluster-wide reset that every node's fold absorbs")
	if err := fs.Parse(args); err != nil {
		return err
	}
	if len(repos) == 0 {
		return fmt.Errorf("reindex wants at least one --repo")
	}
	dir := *data
	if dir == "" {
		for _, p := range repos {
			dir = filepath.Join(p, ".quarry")
			break
		}
	}
	// On external NATS the reset republishes the census to the shared
	// stream, so every node re-folds (idempotent) — reindex is a
	// cluster-wide operation there, not a local-only one. Embedded is the
	// offline single-node rebuild (serve must be stopped: one process per
	// store dir).
	ev, err := startEvents(*natsURL, dir)
	if err != nil {
		return fmt.Errorf("%w (is serve still running against this data dir?)", err)
	}
	defer ev.Close()
	for name, path := range repos {
		if err := os.MkdirAll(filepath.Join(dir, name), 0o755); err != nil {
			return err
		}
		ix, err := index.Open(filepath.Join(dir, name, "index.db"))
		if err != nil {
			return err
		}
		n, err := ingest.NewPipeline(zitcli.New(path), ev, ix, name, "reindex").Reindex()
		ix.Close()
		if err != nil {
			return fmt.Errorf("reindex %s: %w", name, err)
		}
		fmt.Printf("reindexed %s: %d artifacts replayed\n", name, n)
	}
	return nil
}

// repoFlags collects repeatable --repo name=path values (bare paths take
// their directory basename as the repository name).
type repoFlags map[string]string

func (r repoFlags) String() string { return fmt.Sprint(map[string]string(r)) }

func (r repoFlags) Set(v string) error {
	name, path, found := strings.Cut(v, "=")
	if !found {
		path = v
		name = filepath.Base(strings.TrimSuffix(v, "/"))
	}
	if name == "" || path == "" {
		return fmt.Errorf("--repo wants name=path or path, got %q", v)
	}
	if !repoName.MatchString(name) {
		return fmt.Errorf("repository name %q: use letters, digits, dot, dash, underscore", name)
	}
	if _, dup := r[name]; dup {
		return fmt.Errorf("repository %q registered twice", name)
	}
	r[name] = path
	return nil
}

var repoName = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._-]{0,63}$`)

// serveCmd runs the HTTP surface: the op-generated API plus the HTML
// pages, both consumers of one registry. Each --repo gets its own index
// and watcher pipeline; one embedded NATS carries every store's events
// under zit.evt.<name>.* (0001 §4: a watcher per registered store).
func serveCmd(args []string) error {
	fs := flag.NewFlagSet("serve", flag.ContinueOnError)
	addr := fs.String("addr", "127.0.0.1:8080", "listen address")
	repos := repoFlags{}
	fs.Var(repos, "repo", "repository to serve as name=path (repeatable; bare path uses its basename)")
	data := fs.String("data", "", "derived-data directory (default: .quarry beside the first repo)")
	interval := fs.Duration("sync-interval", 2*time.Second, "store poll interval")
	leaseTTL := fs.Duration("lease-ttl", lease.DefaultTTL, "single-writer lease lifetime")
	natsURL := fs.String("nats", "", "external NATS URL for multi-node (default: embedded, single node); the server must allow max_payload >= 8MiB")
	notifyURL := fs.String("notify", "", "notify companion base URL for mentions and activity (default: notifications disabled)")
	if err := fs.Parse(args); err != nil {
		return err
	}

	deps := api.Deps{Version: version, Repos: map[string]*api.Repo{}}
	var notifier web.Notifier
	var chatLive web.ChatLive
	var verify api.Verify
	var fronter *xfer.Fronter
	if len(repos) > 0 {
		fronter = xfer.New()
		defer fronter.Close()
		for name, path := range repos {
			fronter.Register(name, path)
		}
	}
	if len(repos) > 0 {
		dir := *data
		if dir == "" {
			for _, p := range repos {
				dir = filepath.Join(p, ".quarry")
				break
			}
		}
		if err := os.MkdirAll(dir, 0o755); err != nil {
			return err
		}
		odb, err := oltp.Open(filepath.Join(dir, "oltp.db"))
		if err != nil {
			return err
		}
		defer odb.Close()
		fronter.SetAuth(odb)
		// Two credentials, on purpose (0001 §8). Accounts are the forge
		// identity and carry per-repo roles; the RFC §6.3 shared secret
		// stays the SYNC credential, because the protocol fixes its
		// derivation and the server must be able to recompute it. Once
		// any account exists it governs web/API writes; before that,
		// v1's per-repo secret still does, so an existing deployment
		// keeps working and a fresh local one stays open.
		var codeMu sync.Mutex
		codes := map[string]string{}
		verify = func(repo, user, pass string, need ops.Auth) bool {
			if odb.HasAccounts() {
				acct, ok := odb.Authenticate(user, pass)
				if !ok {
					return false
				}
				return odb.RoleOf(acct, repo) >= requiredRole(need)
			}
			if !odb.Gated(repo) {
				return true
			}
			secret, ok := odb.Secret(repo, user)
			if !ok || pass == "" {
				return false
			}
			codeMu.Lock()
			code, cached := codes[repo]
			codeMu.Unlock()
			if !cached {
				path, ok := repos[repo]
				if !ok {
					return false
				}
				c, err := xfer.ProjectCode(path)
				if err != nil {
					log.Printf("auth: project code of %s: %v", repo, err)
					return false
				}
				codeMu.Lock()
				codes[repo], code = c, c
				codeMu.Unlock()
			}
			return secret == xfer.SharedSecret(code, user, pass)
		}

		ev, err := startEvents(*natsURL, dir)
		if err != nil {
			return err
		}
		defer ev.Close()
		notifier = ev

		// Single-writer coordination (0001 §3). node identifies this
		// process; one embedded NATS means one node, but the machinery
		// and its failure posture (coordinator down → writes refuse) run
		// the same on one node as on N.
		leases, err := lease.New(ev.JS(), *addr, *leaseTTL)
		if err != nil {
			return err
		}
		defer leases.Close()
		// Per-repo chat plane (0001 §5): messages fan out over the same
		// NATS the events ride, with a 7-day JetStream window.
		room, err := chat.New(ev.Conn(), ev.JS())
		if err != nil {
			return err
		}
		deps.Chat = room
		chatLive = &chatAdapter{room: room, codes: codes, mu: &codeMu}
		// Notification triggers (0001 §5): fired best-effort to the notify
		// companion; disabled when --notify is unset.
		if *notifyURL != "" {
			deps.Notify = &notifyAdapter{client: notify.New(*notifyURL)}
			fmt.Println("quarry: notifications ->", *notifyURL)
		}
		// Webhook dispatch (Principle 6): always on, because it costs
		// nothing until somebody subscribes — Raise finds no endpoints and
		// returns. Delivery is durable, so an endpoint that is down during
		// a restart still gets its events.
		dispatcher, err := webhook.New(ev.JS(), hookSubs{db: odb})
		if err != nil {
			return err
		}
		hookCtx, stopHooks := context.WithCancel(context.Background())
		defer stopHooks()
		if err := dispatcher.Start(hookCtx); err != nil {
			return err
		}
		deps.Hooks = hookPlane{d: dispatcher, at: func() string {
			return time.Now().UTC().Format(time.RFC3339)
		}}
		// The project codes learned below (shared with the auth verifier's
		// cache) key both the lease and the push guard.
		fronter.SetGuard(func(repo string, proxy func() error) error {
			codeMu.Lock()
			code, ok := codes[repo]
			codeMu.Unlock()
			if !ok {
				return fmt.Errorf("unknown repo %q", repo)
			}
			return leases.Guard(code, proxy)
		})

		for name, path := range repos {
			if err := os.MkdirAll(filepath.Join(dir, name), 0o755); err != nil {
				return err
			}
			ix, err := index.Open(filepath.Join(dir, name, "index.db"))
			if err != nil {
				return err
			}
			defer ix.Close()
			cli := zitcli.New(path)
			// The lease keys on the store's project code, learned via the
			// same boundary-clean handshake user-add uses.
			code, cerr := xfer.ProjectCode(path)
			if cerr != nil {
				return fmt.Errorf("project code of %s: %w", name, cerr)
			}
			codeMu.Lock()
			codes[name] = code
			codeMu.Unlock()
			// Flips false when the first ingest finishes; the fleet page
			// shows "indexing…" until then so a zero count is not
			// mistaken for an empty repository.
			indexing := new(atomic.Bool)
			indexing.Store(true)
			deps.Repos[name] = &api.Repo{
				Index: ix, Zit: cli, Code: code,
				Indexing: indexing.Load,
				// A lease refusal ("not the writer" / coordinator down)
				// surfaces as a 503 the client can retry, not a 500.
				Guard: func(fn func() error) error {
					err := leases.Guard(code, fn)
					if errors.Is(err, lease.ErrNotHolder) || errors.Is(err, lease.ErrUnavailable) {
						return fmt.Errorf("%w: %v", ops.ErrUnavailable, err)
					}
					return err
				},
			}

			pipe := ingest.NewPipeline(cli, ev, ix, name, *addr)
			deps.Repos[name].Sync = pipe.Sync
			// Ingest runs in the background so the server listens
			// immediately. A large store's first fold takes minutes
			// (sqlite.org's repo: 131k artifacts, ~9 min), and blocking
			// the listener on it makes every other repo unreachable and
			// makes a working server look dead. Repos come online as
			// they finish; until then a repo reads as empty, which is
			// honest — the index is derived state catching up.
			go func(name, path string, pipe *ingest.Pipeline, indexing *atomic.Bool) {
				start := time.Now()
				n, err := pipe.Sync()
				indexing.Store(false)
				if err != nil {
					log.Printf("initial ingest of %s: %v", name, err)
				} else {
					log.Printf("ingested %d artifacts from %s (%s) in %s",
						n, name, path, time.Since(start).Round(time.Second))
				}
				for range time.Tick(*interval) {
					if _, err := pipe.Sync(); err != nil {
						log.Printf("ingest %s: %v", name, err)
					}
				}
			}(name, path, pipe, indexing)
		}
	}

	mux, err := buildHandler(deps, notifier, chatLive, verify)
	if err != nil {
		return err
	}
	if fronter != nil {
		mount := http.NewServeMux()
		// Sync fronting (0001 §7): zit clients POST the repo URL
		// verbatim; /xfer is the alias zit serve itself answers on.
		mount.Handle("POST /r/{repo}", fronter)
		mount.Handle("POST /r/{repo}/xfer", fronter)
		mount.Handle("/", mux)
		mux = mount
	}
	fmt.Println("quarry", version, "listening on", *addr)
	return http.ListenAndServe(*addr, mux)
}

// chatAdapter bridges the chat data plane to web.ChatLive: it maps a
// repo name to its project code and converts messages to the view type,
// keeping web free of the data-plane package.
type chatAdapter struct {
	room  *chat.Room
	codes map[string]string
	mu    *sync.Mutex
}

func (a *chatAdapter) SubscribeChat(repo string) (<-chan web.ChatMessage, func()) {
	a.mu.Lock()
	code, ok := a.codes[repo]
	a.mu.Unlock()
	if !ok {
		ch := make(chan web.ChatMessage)
		close(ch)
		return ch, func() {}
	}
	src, cancel := a.room.Subscribe(code)
	out := make(chan web.ChatMessage, 32)
	go func() {
		defer close(out)
		for m := range src {
			select {
			case out <- web.ChatMessage{User: m.User, Text: m.Text, At: m.At}:
			default:
			}
		}
	}()
	return out, cancel
}

// notifyAdapter bridges api.NotifyPlane to the notify client, mapping a
// quarry event to a Novu-shaped trigger. The event's To is the
// subscriber id (a user or a "repo:<name>" topic key).
type notifyAdapter struct{ client *notify.Client }

func (a *notifyAdapter) Trigger(ctx context.Context, e api.NotifyEvent) error {
	return a.client.Trigger(ctx, e.Workflow,
		notify.Recipient{SubscriberID: e.To}, e.Payload, e.TxnID)
}

// startEvents opens the event plane: an external NATS when a URL is
// given (multi-node), else an embedded server under dir (single node).
func startEvents(url, dir string) (*events.Server, error) {
	if url != "" {
		fmt.Println("quarry: connecting to external NATS", url)
		return events.Connect(url)
	}
	return events.Start(filepath.Join(dir, "nats"))
}

// buildHandler wires the two registry consumers onto one mux. notifier
// may be nil (no live pages).
func buildHandler(deps api.Deps, notifier web.Notifier, chatLive web.ChatLive, verify api.Verify) (http.Handler, error) {
	r := ops.NewRegistry()
	if err := api.RegisterAll(r, deps); err != nil {
		return nil, err
	}
	apiRouter, err := api.Router(r, deps.Version, verify)
	if err != nil {
		return nil, err
	}
	mux := http.NewServeMux()
	mux.Handle("/api/", apiRouter)
	mux.Handle("/", web.Handler(r, notifier, chatLive, web.Verify(verify)))
	return mux, nil
}

// printSpec emits the OpenAPI document generated from the full registry —
// the same registry serve will run, so the spec cannot drift (0001 §9).
func printSpec(w *os.File) error {
	r := ops.NewRegistry()
	if err := api.RegisterAll(r, api.Deps{Version: version}); err != nil {
		return err
	}
	spec, err := ops.Spec(r, "quarry", version)
	if err != nil {
		return err
	}
	enc := json.NewEncoder(w)
	enc.SetIndent("", "  ")
	return enc.Encode(spec)
}

// requiredRole maps an op's declared capability onto the role an account
// must hold on that repo. This is the only place the ops vocabulary and
// the ACL vocabulary meet, so adding a level is a change here and
// nowhere else.
func requiredRole(need ops.Auth) oltp.Role {
	switch need {
	case ops.Admin:
		return oltp.RepoAdmin
	case ops.Token:
		return oltp.Write
	}
	return oltp.NoRole
}

// accountCmd manages forge accounts and per-repo roles (0001 §8's v2
// tables). Distinct from `quarry user add`, which writes the SYNC
// credential: that one is the RFC §6.3 shared secret a push presents,
// this one is the login a person uses. --repo-name names a REGISTERED
// repository, because roles are held against served names, not paths.
func accountCmd(args []string) error {
	const usage = `usage:
  quarry account add    --user U [--password P] [--admin]
  quarry account passwd --user U [--password P]
  quarry account grant  --user U --repo-name N --role read|write|admin
  quarry account revoke --user U --repo-name N
  quarry account delete --user U
  quarry account list
every form also takes --data DIR, or one --repo name=path to locate it`
	if len(args) < 1 {
		return errors.New(usage)
	}
	sub := args[0]
	fs := flag.NewFlagSet("account "+sub, flag.ContinueOnError)
	repos := repoFlags{}
	fs.Var(repos, "repo", "a registered repository as name=path, used only to locate the data directory")
	data := fs.String("data", "", "derived-data directory (default: .quarry beside the repo)")
	user := fs.String("user", "", "account name")
	password := fs.String("password", "", "password (omit to read from QUARRY_PASSWORD)")
	repoTarget := fs.String("repo-name", "", "repository the role applies to")
	role := fs.String("role", "", "read, write or admin")
	admin := fs.Bool("admin", false, "site admin: every repository, every capability")
	if err := fs.Parse(args[1:]); err != nil {
		return err
	}

	dir := *data
	if dir == "" {
		if len(repos) != 1 {
			return errors.New("need --data DIR, or one --repo name=path to locate it\n" + usage)
		}
		for _, p := range repos {
			dir = filepath.Join(p, ".quarry")
		}
	}
	db, err := oltp.Open(filepath.Join(dir, "oltp.db"))
	if err != nil {
		return err
	}
	defer db.Close()

	// Password sources in order: the flag, then the environment, so a
	// password need not reach the process table or shell history.
	readPassword := func() (string, error) {
		if *password != "" {
			return *password, nil
		}
		if pw := os.Getenv("QUARRY_PASSWORD"); pw != "" {
			return pw, nil
		}
		return "", errors.New("no password: use --password or QUARRY_PASSWORD")
	}

	switch sub {
	case "add":
		if *user == "" {
			return errors.New("account add wants --user")
		}
		pw, err := readPassword()
		if err != nil {
			return err
		}
		first := !db.HasAccounts()
		if err := db.AddAccount(*user, pw, *admin); err != nil {
			return err
		}
		if *admin {
			fmt.Printf("account %q created (site admin)\n", *user)
		} else {
			fmt.Printf("account %q created\n", *user)
		}
		if first {
			fmt.Println("this is the first account: web and API writes now require one")
		}
		return nil

	case "passwd":
		if *user == "" {
			return errors.New("account passwd wants --user")
		}
		pw, err := readPassword()
		if err != nil {
			return err
		}
		if err := db.SetPassword(*user, pw); err != nil {
			return err
		}
		fmt.Printf("password changed for %q\n", *user)
		return nil

	case "grant":
		if *user == "" || *repoTarget == "" || *role == "" {
			return errors.New("account grant wants --user, --repo-name and --role")
		}
		r, err := oltp.ParseRole(*role)
		if err != nil {
			return err
		}
		if err := db.Grant(*user, *repoTarget, r); err != nil {
			return err
		}
		fmt.Printf("%q now has %s on %s\n", *user, r, *repoTarget)
		return nil

	case "revoke":
		if *user == "" || *repoTarget == "" {
			return errors.New("account revoke wants --user and --repo-name")
		}
		if err := db.Revoke(*user, *repoTarget); err != nil {
			return err
		}
		fmt.Printf("%q no longer has a role on %s\n", *user, *repoTarget)
		return nil

	case "delete":
		if *user == "" {
			return errors.New("account delete wants --user")
		}
		if err := db.DeleteAccount(*user); err != nil {
			return err
		}
		fmt.Printf("account %q deleted, with every role it held\n", *user)
		return nil

	case "list":
		accts, err := db.Accounts()
		if err != nil {
			return err
		}
		if len(accts) == 0 {
			fmt.Println("no accounts: web and API writes are open (single-user local serving)")
			return nil
		}
		for _, a := range accts {
			line := a.Name
			if a.Admin {
				line += " (site admin)"
			}
			grants, err := db.GrantsFor(a.Name)
			if err != nil {
				return err
			}
			for _, repo := range oltp.SortedRepos(grants) {
				line += fmt.Sprintf("  %s=%s", repo, grants[repo])
			}
			fmt.Println(line)
		}
		return nil
	}
	return errors.New(usage)
}

// hookSubs adapts the OLTP database to the dispatcher's Subscriptions
// view. The two packages keep their own types on purpose: oltp owns the
// stored row (secrets, timestamps, paused flag), webhook owns what a
// delivery needs, and this is the only place they meet.
type hookSubs struct{ db *oltp.DB }

func (h hookSubs) ActiveWebhooks(repo string) ([]webhook.Sub, error) {
	rows, err := h.db.ActiveWebhooks(repo)
	if err != nil {
		return nil, err
	}
	out := make([]webhook.Sub, 0, len(rows))
	for _, w := range rows {
		out = append(out, webhook.Sub{ID: w.ID, URL: w.URL, Secret: w.Secret, Active: true})
	}
	return out, nil
}

func (h hookSubs) SubByID(id int64) (webhook.Sub, bool) {
	w, ok := h.db.WebhookByID(id)
	if !ok {
		return webhook.Sub{}, false
	}
	return webhook.Sub{ID: w.ID, URL: w.URL, Secret: w.Secret, Active: w.Active}, true
}

// hookPlane adapts the dispatcher to the ops layer's WebhookPlane, which
// speaks primitives so the ops never import the delivery machinery.
type hookPlane struct {
	d  *webhook.Dispatcher
	at func() string
}

func (h hookPlane) Raise(id, repo, kind, actor, summary, link string) error {
	return h.d.Raise(webhook.Event{
		ID: id, Repo: repo, Kind: kind, Actor: actor,
		Summary: summary, Link: link, At: h.at(),
	})
}

// webhookCmd manages subscriptions to quarry's outbound event surface
// (Principle 6). The signing secret is printed once at creation and
// never again — a key a listing can read back is a key that leaks
// through logs, screenshots and shoulders.
func webhookCmd(args []string) error {
	const usage = `usage:
  quarry webhook add    --repo-name N --url https://...
  quarry webhook list   --repo-name N
  quarry webhook pause  --id N
  quarry webhook resume --id N
  quarry webhook delete --id N
every form also takes --data DIR, or one --repo name=path to locate it`
	if len(args) < 1 {
		return errors.New(usage)
	}
	sub := args[0]
	fs := flag.NewFlagSet("webhook "+sub, flag.ContinueOnError)
	repos := repoFlags{}
	fs.Var(repos, "repo", "a registered repository as name=path, used only to locate the data directory")
	data := fs.String("data", "", "derived-data directory")
	repoTarget := fs.String("repo-name", "", "repository whose events to deliver")
	endpoint := fs.String("url", "", "endpoint to POST events to")
	id := fs.Int64("id", 0, "subscription id")
	if err := fs.Parse(args[1:]); err != nil {
		return err
	}
	dir := *data
	if dir == "" {
		if len(repos) != 1 {
			return errors.New("need --data DIR, or one --repo name=path to locate it\n" + usage)
		}
		for _, p := range repos {
			dir = filepath.Join(p, ".quarry")
		}
	}
	db, err := oltp.Open(filepath.Join(dir, "oltp.db"))
	if err != nil {
		return err
	}
	defer db.Close()

	switch sub {
	case "add":
		if *repoTarget == "" || *endpoint == "" {
			return errors.New("webhook add wants --repo-name and --url")
		}
		w, err := db.AddWebhook(*repoTarget, *endpoint)
		if err != nil {
			return err
		}
		fmt.Printf("webhook %d: %s -> %s\n", w.ID, w.Repo, w.URL)
		fmt.Printf("signing secret (shown once): %s\n", w.Secret)
		fmt.Println("verify X-Quarry-Signature as sha256=HMAC-SHA256(secret, body)")
		return nil
	case "list":
		if *repoTarget == "" {
			return errors.New("webhook list wants --repo-name")
		}
		hooks, err := db.Webhooks(*repoTarget)
		if err != nil {
			return err
		}
		if len(hooks) == 0 {
			fmt.Printf("no webhooks on %s\n", *repoTarget)
			return nil
		}
		for _, w := range hooks {
			state := "active"
			if !w.Active {
				state = "paused"
			}
			fmt.Printf("%d  %-7s %s  (added %s)\n", w.ID, state, w.URL, w.Created)
		}
		return nil
	case "pause", "resume":
		if *id == 0 {
			return fmt.Errorf("webhook %s wants --id", sub)
		}
		if err := db.SetWebhookActive(*id, sub == "resume"); err != nil {
			return err
		}
		fmt.Printf("webhook %d %sd\n", *id, sub)
		return nil
	case "delete":
		if *id == 0 {
			return errors.New("webhook delete wants --id")
		}
		if err := db.DeleteWebhook(*id); err != nil {
			return err
		}
		fmt.Printf("webhook %d deleted\n", *id)
		return nil
	}
	return errors.New(usage)
}