indexer/store/store.go

// Package store provides durable persistence of scraped projects in SQLite.
//
// The database is the system of record: scrapers write discovered projects
// here, and a search index (e.g. Typesense) can be built or rebuilt from it
// without re-scraping the forges. Projects are stored as one row each, with
// the URL as the natural key so re-running the scrape upserts in place.
package store

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	_ "modernc.org/sqlite"

	"git.xengi.de/ffd/indexer/scraper"
)

const schema = `
CREATE TABLE IF NOT EXISTS projects (
	url                 TEXT PRIMARY KEY,
	name                TEXT NOT NULL,
	description         TEXT NOT NULL DEFAULT '',
	topics              TEXT NOT NULL DEFAULT '[]',   -- JSON array
	languages           TEXT NOT NULL DEFAULT '[]',   -- JSON array
	licenses            TEXT NOT NULL DEFAULT '[]',   -- JSON array
	open_issues         INTEGER NOT NULL DEFAULT 0,
	open_prs            INTEGER NOT NULL DEFAULT 0,
	latest_commit       INTEGER NOT NULL DEFAULT 0,   -- unix epoch seconds (UTC)
	latest_release_name TEXT NOT NULL DEFAULT '',
	latest_release_date INTEGER NOT NULL DEFAULT 0,   -- unix epoch seconds (UTC)
	readme              TEXT NOT NULL DEFAULT '',
	uses_ai             INTEGER NOT NULL DEFAULT 0,
	scraped_at          INTEGER NOT NULL DEFAULT 0,   -- unix epoch seconds (UTC)
	forge_url           TEXT NOT NULL DEFAULT '',
	forge_type          TEXT NOT NULL DEFAULT ''
);

CREATE TABLE IF NOT EXISTS forge_status (
	url               TEXT PRIMARY KEY,
	forge_type        TEXT NOT NULL DEFAULT '',
	version           TEXT NOT NULL DEFAULT '',   -- forge software version, e.g. 16.0.5
	online            INTEGER NOT NULL DEFAULT 0,
	last_scraped_at   INTEGER NOT NULL DEFAULT 0,   -- unix epoch seconds (UTC)
	last_error        TEXT NOT NULL DEFAULT ''
);
`

// schemaMigrations adds columns introduced after the initial schema. They are
// applied only if the column does not already exist, so both fresh databases
// (whose CREATE TABLE already ships the columns) and old databases upgrade
// cleanly.
var schemaMigrations = []string{
	// forge identity per project, used by the metrics daemon to label per-repo
	// series and to count repositories per forge.
	`ALTER TABLE projects ADD COLUMN forge_url TEXT NOT NULL DEFAULT ''`,
	`ALTER TABLE projects ADD COLUMN forge_type TEXT NOT NULL DEFAULT ''`,
	// forge software version, detected at scrape time and exposed as a metric.
	`ALTER TABLE forge_status ADD COLUMN version TEXT NOT NULL DEFAULT ''`,
}

// schemaColumnDrops removes columns that are no longer wanted. They are
// applied only if the column exists.
var schemaColumnDrops = []string{
	// forge name was removed: the URL is the forge's unique identifier, and
	// there is no reliable way to derive a display name across forge types.
	`ALTER TABLE projects DROP COLUMN forge_name`,
	`ALTER TABLE forge_status DROP COLUMN name`,
}

// Store wraps a SQLite database holding scraped projects.
type Store struct {
	db *sql.DB
}

// Open opens (creating if needed) the SQLite database at path.
func Open(path string) (*Store, error) {
	db, err := sql.Open("sqlite", path)
	if err != nil {
		return nil, fmt.Errorf("open sqlite db %q: %w", path, err)
	}
	if _, err := db.Exec(schema); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("create schema: %w", err)
	}
	if err := applyMigrations(db); err != nil {
		_ = db.Close()
		return nil, err
	}
	return &Store{db: db}, nil
}

// applyMigrations runs schemaMigrations (add columns if missing) and
// schemaColumnDrops (drop columns if present).
func applyMigrations(db *sql.DB) error {
	for _, m := range schemaMigrations {
		table, col := alterColumn(m)
		has, err := hasColumn(db, table, col)
		if err != nil {
			return fmt.Errorf("check column %q: %w", col, err)
		}
		if has {
			continue
		}
		if _, err := db.Exec(m); err != nil {
			return fmt.Errorf("apply migration %q: %w", m, err)
		}
	}

	for _, d := range schemaColumnDrops {
		table, col := dropColumn(d)
		has, err := hasColumn(db, table, col)
		if err != nil {
			return fmt.Errorf("check column %q: %w", col, err)
		}
		if !has {
			continue
		}
		if _, err := db.Exec(d); err != nil {
			return fmt.Errorf("apply column drop %q: %w", d, err)
		}
	}
	return nil
}

// alterColumn extracts the table and column name from an
// "ALTER TABLE <table> ADD COLUMN <name> <type>" statement.
func alterColumn(stmt string) (string, string) {
	t := strings.TrimPrefix(stmt, "ALTER TABLE ")
	table, rest, ok := strings.Cut(t, " ADD COLUMN ")
	if !ok {
		return "", ""
	}
	col := rest
	if i := strings.IndexByte(col, ' '); i >= 0 {
		col = col[:i]
	}
	return table, col
}

// dropColumn extracts the table and column name from an
// "ALTER TABLE <table> DROP COLUMN <column>" statement.
func dropColumn(stmt string) (string, string) {
	t := strings.TrimPrefix(stmt, "ALTER TABLE ")
	table, rest, ok := strings.Cut(t, " DROP COLUMN ")
	if !ok {
		return "", ""
	}
	return table, rest
}

// hasColumn reports whether table has a column with the given name.
func hasColumn(db *sql.DB, table, column string) (bool, error) {
	rows, err := db.Query(`PRAGMA table_info(` + table + `)`)
	if err != nil {
		return false, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var (
			cid     int
			name    string
			ctype   string
			notnull int
			dflt    sql.NullString
			pk      int
		)
		if err := rows.Scan(&cid, &name, &ctype, &notnull, &dflt, &pk); err != nil {
			return false, err
		}
		if name == column {
			return true, nil
		}
	}
	return false, rows.Err()
}

// Close closes the underlying database.
func (s *Store) Close() error {
	return s.db.Close()
}

const upsertSQL = `
INSERT INTO projects (
	url, name, description, topics, languages, licenses,
	open_issues, open_prs, latest_commit,
	latest_release_name, latest_release_date,
	readme, uses_ai, scraped_at,
	forge_url, forge_type
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(url) DO UPDATE SET
	name = excluded.name,
	description = excluded.description,
	topics = excluded.topics,
	languages = excluded.languages,
	licenses = excluded.licenses,
	open_issues = excluded.open_issues,
	open_prs = excluded.open_prs,
	latest_commit = excluded.latest_commit,
	latest_release_name = excluded.latest_release_name,
	latest_release_date = excluded.latest_release_date,
	readme = excluded.readme,
	uses_ai = excluded.uses_ai,
	scraped_at = excluded.scraped_at,
	forge_url = excluded.forge_url,
	forge_type = excluded.forge_type;
`

// Upsert inserts or updates a project, keyed on its URL.
func (s *Store) Upsert(p scraper.Project) error {
	topics, err := json.Marshal(p.Topics)
	if err != nil {
		return fmt.Errorf("marshal topics: %w", err)
	}
	langs, err := json.Marshal(p.Languages)
	if err != nil {
		return fmt.Errorf("marshal languages: %w", err)
	}
	licenses, err := json.Marshal(p.Licenses)
	if err != nil {
		return fmt.Errorf("marshal licenses: %w", err)
	}

	releaseDate := timeToUnix(p.LatestRelease.Date)
	latestCommit := timeToUnix(p.LatestCommit)

	_, err = s.db.Exec(upsertSQL,
		p.Url,
		p.Name,
		p.Description,
		string(topics),
		string(langs),
		string(licenses),
		p.OpenIssues,
		p.OpenPRs,
		latestCommit,
		p.LatestRelease.Name,
		releaseDate,
		p.Readme,
		boolToInt(p.UsesAI),
		time.Now().UTC().Unix(),
		p.ForgeURL,
		p.ForgeType,
	)
	return err
}

// All returns every stored project.
func (s *Store) All() ([]scraper.Project, error) {
	rows, err := s.db.Query(`SELECT
		url, name, description, topics, languages, licenses,
		open_issues, open_prs, latest_commit,
		latest_release_name, latest_release_date,
		readme, uses_ai, forge_url, forge_type
		FROM projects`)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	var projects []scraper.Project
	for rows.Next() {
		var (
			p            scraper.Project
			topicsJSON   string
			langsJSON    string
			licensesJSON string
			latestCommit int64
			relName      string
			relDate      int64
			usesAI       int
		)
		if err := rows.Scan(
			&p.Url, &p.Name, &p.Description,
			&topicsJSON, &langsJSON, &licensesJSON,
			&p.OpenIssues, &p.OpenPRs, &latestCommit,
			&relName, &relDate,
			&p.Readme, &usesAI, &p.ForgeURL, &p.ForgeType,
		); err != nil {
			return nil, err
		}

		p.UsesAI = usesAI != 0
		if err := json.Unmarshal([]byte(topicsJSON), &p.Topics); err != nil {
			return nil, err
		}
		if err := json.Unmarshal([]byte(langsJSON), &p.Languages); err != nil {
			return nil, err
		}
		if err := json.Unmarshal([]byte(licensesJSON), &p.Licenses); err != nil {
			return nil, err
		}
		p.LatestCommit = time.Unix(latestCommit, 0).UTC()
		p.LatestRelease.Name = relName
		p.LatestRelease.Date = time.Unix(relDate, 0).UTC()

		projects = append(projects, p)
	}
	return projects, rows.Err()
}

// timeToUnix returns the Unix epoch seconds for a time, or 0 for the zero
// time (which is stored as a 0/absent value).
func timeToUnix(t time.Time) int64 {
	if t.IsZero() {
		return 0
	}
	return t.Unix()
}

func boolToInt(b bool) int {
	if b {
		return 1
	}
	return 0
}

// ForgeStatus is the per-forge state recorded on every scrape run (success or
// failure), consumed by the metrics daemon. The forge URL is the unique
// identifier.
type ForgeStatus struct {
	URL           string
	ForgeType     string
	Version       string
	Online        bool
	LastScrapedAt time.Time
	LastError     string
}

// RecordForgeStatus upserts the scrape result for a single forge, keyed on
// its URL. It is written even when the forge failed to scrape (Online=false,
// LastError set) so that the metrics daemon always sees a row per forge.
func (s *Store) RecordForgeStatus(st ForgeStatus) error {
	_, err := s.db.Exec(`INSERT INTO forge_status (
		url, forge_type, version, online, last_scraped_at, last_error
	) VALUES (?, ?, ?, ?, ?, ?)
	ON CONFLICT(url) DO UPDATE SET
		forge_type = excluded.forge_type,
		version = excluded.version,
		online = excluded.online,
		last_scraped_at = excluded.last_scraped_at,
		last_error = excluded.last_error;`,
		st.URL,
		st.ForgeType,
		st.Version,
		boolToInt(st.Online),
		st.LastScrapedAt.UTC().Unix(),
		st.LastError,
	)
	return err
}

// ForgeStatuses returns every recorded forge status row.
func (s *Store) ForgeStatuses() ([]ForgeStatus, error) {
	rows, err := s.db.Query(`SELECT
		url, forge_type, version, online, last_scraped_at, last_error
		FROM forge_status
		ORDER BY url`)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	var statuses []ForgeStatus
	for rows.Next() {
		var (
			st     ForgeStatus
			online int
			unix   int64
		)
		if err := rows.Scan(&st.URL, &st.ForgeType, &st.Version, &online, &unix, &st.LastError); err != nil {
			return nil, err
		}
		st.Online = online != 0
		st.LastScrapedAt = time.Unix(unix, 0).UTC()
		statuses = append(statuses, st)
	}
	return statuses, rows.Err()
}

// ForgeStatus returns the status row for a single forge URL, or zero value if
// there is none yet. It is used to fetch the last detected forge type and
// version.
func (s *Store) ForgeStatus(url string) (ForgeStatus, error) {
	var (
		st     ForgeStatus
		online int
		unix   int64
	)
	err := s.db.QueryRow(`SELECT
		url, forge_type, version, online, last_scraped_at, last_error
		FROM forge_status
		WHERE url = ?`, url).Scan(&st.URL, &st.ForgeType, &st.Version, &online, &unix, &st.LastError)
	if err == sql.ErrNoRows {
		return ForgeStatus{}, nil
	}
	if err != nil {
		return ForgeStatus{}, err
	}
	st.Online = online != 0
	st.LastScrapedAt = time.Unix(unix, 0).UTC()
	return st, nil
}

// RepoIdentity is the subset of a project needed by the metrics daemon: its
// forge identity and language list. The full Project has the same fields; the
// dedicated type keeps All() unencumbered and makes the metrics intent clear.
type RepoIdentity struct {
	ForgeURL  string
	ForgeType string
	URL       string
	Languages []string
}

// RepoIdentities returns every stored project's forge identity and languages,
// used by the metrics daemon.
func (s *Store) RepoIdentities() ([]RepoIdentity, error) {
	rows, err := s.db.Query(`SELECT
		url, languages, forge_url, forge_type
		FROM projects
		ORDER BY url`)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	var repos []RepoIdentity
	for rows.Next() {
		var (
			r         RepoIdentity
			langsJSON string
		)
		if err := rows.Scan(&r.URL, &langsJSON, &r.ForgeURL, &r.ForgeType); err != nil {
			return nil, err
		}
		if err := json.Unmarshal([]byte(langsJSON), &r.Languages); err != nil {
			return nil, err
		}
		repos = append(repos, r)
	}
	return repos, rows.Err()
}