Skip to content
quarry Artifact [0e1f114c8b]
Login

Artifact [0e1f114c8b]

Check-ins Using Download Hex

Artifact 0e1f114c8ba0e18b9ed8dbcf4c6609a359e9384999889ab986fa9dfb391ff914:

// mirror.go stores which upstream each served store tracks and how
// often quarry pulls from it. A mirror is forge configuration, not DAG
// state, so it belongs in the OLTP database beside accounts and webhooks
// (0001 §8) rather than in the disposable index.
//
// The remote deliberately does NOT live in the store via `zit remote`.
// That would put the same fact in two places — quarry's schedule here,
// zit's URL there — with no mechanism to keep them agreeing, and the
// failure mode is a store quietly mirroring somewhere the operator did
// not configure.
package oltp

import (
	"database/sql"
	"errors"
	"fmt"
	"net/url"
	"time"
)

const mirrorSchema = `
CREATE TABLE IF NOT EXISTS mirror (
	repo     TEXT PRIMARY KEY,        -- the forge's name for the store
	url      TEXT NOT NULL,           -- upstream sync endpoint
	every    INTEGER NOT NULL,        -- seconds between pulls
	enabled  INTEGER NOT NULL DEFAULT 1,
	last_at  TEXT NOT NULL DEFAULT '',-- last ATTEMPT, success or not
	last_ok  TEXT NOT NULL DEFAULT '',-- last success
	last_err TEXT NOT NULL DEFAULT '',
	failures INTEGER NOT NULL DEFAULT 0
);`

// MinMirrorInterval floors how often quarry will pull. Upstreams here
// are other people's servers: fossil-scm.org, core.tcl-lang.org,
// sqlite.org. A misconfigured schedule is not quarry's problem to
// absorb, it is theirs to absorb, so the floor is a courtesy that the
// configuration surface cannot opt out of.
const MinMirrorInterval = 60 * time.Second

// Mirror is one store's upstream tracking configuration and its recent
// history.
type Mirror struct {
	Repo    string
	URL     string
	Every   time.Duration
	Enabled bool
	// LastAt is the last attempt, LastOK the last success — kept apart
	// so "tried and failed" is distinguishable from "never due yet".
	LastAt   string
	LastOK   string
	LastErr  string
	Failures int
}

// Due reports whether this mirror should be pulled now. A mirror that
// has never run is due immediately; a disabled one never is.
func (m Mirror) Due(now time.Time) bool {
	if !m.Enabled {
		return false
	}
	if m.LastAt == "" {
		return true
	}
	last, err := time.Parse(time.RFC3339, m.LastAt)
	if err != nil {
		// An unparseable stamp is corruption, not a reason to stop
		// mirroring — treat it as "overdue" and let the next write fix it.
		return true
	}
	return now.Sub(last) >= m.Every
}

// SetMirror configures (or reconfigures) a store's upstream. Reconfiguring
// preserves the history: an operator correcting a typo in the URL has not
// made the previous failures untrue.
func (d *DB) SetMirror(repo, endpoint string, every time.Duration) error {
	if repo == "" {
		return errors.New("oltp: mirror needs a repository")
	}
	// Only http(s), for the same reason a webhook URL is restricted: the
	// server fetches this on its own initiative, so any other scheme is a
	// request to make quarry speak a protocol on someone else's behalf.
	u, err := url.Parse(endpoint)
	if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
		return fmt.Errorf("oltp: mirror url must be http or https, got %q", endpoint)
	}
	if every < MinMirrorInterval {
		every = MinMirrorInterval
	}
	_, err = d.db.Exec(`
		INSERT INTO mirror (repo, url, every, enabled) VALUES (?, ?, ?, 1)
		ON CONFLICT(repo) DO UPDATE SET url = excluded.url, every = excluded.every, enabled = 1`,
		repo, endpoint, int64(every.Seconds()))
	return err
}

// DisableMirror stops scheduled pulls without forgetting the URL, so
// re-enabling does not mean retyping it.
func (d *DB) DisableMirror(repo string) error {
	_, err := d.db.Exec(`UPDATE mirror SET enabled = 0 WHERE repo = ?`, repo)
	return err
}

// DeleteMirror forgets a store's upstream entirely.
func (d *DB) DeleteMirror(repo string) error {
	_, err := d.db.Exec(`DELETE FROM mirror WHERE repo = ?`, repo)
	return err
}

// Mirrors lists every configured mirror, repo-ordered.
func (d *DB) Mirrors() ([]Mirror, error) {
	rows, err := d.db.Query(`
		SELECT repo, url, every, enabled, last_at, last_ok, last_err, failures
		FROM mirror ORDER BY repo`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Mirror
	for rows.Next() {
		var m Mirror
		var every int64
		var enabled int
		if err := rows.Scan(&m.Repo, &m.URL, &every, &enabled,
			&m.LastAt, &m.LastOK, &m.LastErr, &m.Failures); err != nil {
			return nil, err
		}
		m.Every = time.Duration(every) * time.Second
		m.Enabled = enabled != 0
		out = append(out, m)
	}
	return out, rows.Err()
}

// MirrorFor returns one store's mirror, if it has one.
func (d *DB) MirrorFor(repo string) (Mirror, bool, error) {
	var m Mirror
	var every int64
	var enabled int
	err := d.db.QueryRow(`
		SELECT repo, url, every, enabled, last_at, last_ok, last_err, failures
		FROM mirror WHERE repo = ?`, repo).
		Scan(&m.Repo, &m.URL, &every, &enabled,
			&m.LastAt, &m.LastOK, &m.LastErr, &m.Failures)
	if errors.Is(err, sql.ErrNoRows) {
		return Mirror{}, false, nil
	}
	if err != nil {
		return Mirror{}, false, err
	}
	m.Every = time.Duration(every) * time.Second
	m.Enabled = enabled != 0
	return m, true, nil
}

// RecordMirrorRun stamps an attempt. A success clears the error and the
// failure streak; a failure leaves last_ok alone, so "last succeeded"
// stays answerable however long a store has been broken.
func (d *DB) RecordMirrorRun(repo string, at time.Time, runErr error) error {
	stamp := at.UTC().Format(time.RFC3339)
	if runErr != nil {
		_, err := d.db.Exec(`
			UPDATE mirror SET last_at = ?, last_err = ?, failures = failures + 1
			WHERE repo = ?`, stamp, runErr.Error(), repo)
		return err
	}
	_, err := d.db.Exec(`
		UPDATE mirror SET last_at = ?, last_ok = ?, last_err = '', failures = 0
		WHERE repo = ?`, stamp, stamp, repo)
	return err
}