Skip to content
quarry Artifact [e93e59af61]
Login

Artifact [e93e59af61]

Check-ins Using Download Hex

Artifact e93e59af61496b1cbb735bc2029c0c25f71f1fe641416e5e39e788cbeb2a8494:

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

import (
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"strings"
	"sync"
	"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/oltp"
	"craftdesign.group/quarry/internal/ops"
	"craftdesign.group/quarry/internal/web"
	"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
		}
	}
	fmt.Fprintln(os.Stderr, "usage: quarry version|spec|serve|reindex|user|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")
	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)
		// One credential story: web/API writes verify HTTP Basic against
		// the same RFC §6.3 shared secrets sync pushes use. Project
		// codes resolve lazily, once per repo.
		var codeMu sync.Mutex
		codes := map[string]string{}
		verify = func(repo, user, pass string) bool {
			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}
		// 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()
			deps.Repos[name] = &api.Repo{
				Index: ix, Zit: cli, Code: code,
				// 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
			if n, err := pipe.Sync(); err != nil {
				return fmt.Errorf("initial ingest of %s: %w", name, err)
			} else {
				fmt.Printf("ingested %d artifacts from %s (%s)\n", n, name, path)
			}
			go func(name string, pipe *ingest.Pipeline) {
				for range time.Tick(*interval) {
					if _, err := pipe.Sync(); err != nil {
						log.Printf("ingest %s: %v", name, err)
					}
				}
			}(name, pipe)
		}
	}

	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
}

// 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)
}