package cache
import (
"context"
"database/sql"
"errors"
"fmt"
)
const currentSchemaVersion = 21
const issueCommentSyncSchemaVersion = 16
func CurrentSchemaVersion() int {
return currentSchemaVersion
}
func IssueCommentSyncSchemaVersion() int {
return issueCommentSyncSchemaVersion
}
type VersionCompatibility struct {
DetectedVersion int
ExpectedVersion int
Compatible bool
PermitWrites bool
Message string
Remediation string
}
var ErrSchemaVersionIncompatible = errors.New("cache: schema version is incompatible")
type SchemaVersionError struct {
Compat VersionCompatibility
}
func (e *SchemaVersionError) Error() string {
if e == nil {
return ErrSchemaVersionIncompatible.Error()
}
if e.Compat.Remediation == "" {
return fmt.Sprintf("%s: detected=%d expected=%d: %s", ErrSchemaVersionIncompatible, e.Compat.DetectedVersion, e.Compat.ExpectedVersion, e.Compat.Message)
}
return fmt.Sprintf("%s: detected=%d expected=%d: %s; %s", ErrSchemaVersionIncompatible, e.Compat.DetectedVersion, e.Compat.ExpectedVersion, e.Compat.Message, e.Compat.Remediation)
}
func (e *SchemaVersionError) Unwrap() error {
return ErrSchemaVersionIncompatible
}
func (e *SchemaVersionError) DiagnosticCode() string { return "cache_schema_blocked" }
func CheckVersionCompatibility(ctx context.Context, db *sql.DB) (VersionCompatibility, error) {
expected := currentSchemaVersion
hasTable, err := hasSchemaVersionTable(ctx, db)
if err != nil {
return VersionCompatibility{}, err
}
if !hasTable {
empty, err := isEmptyDatabase(ctx, db)
if err != nil {
return VersionCompatibility{}, err
}
if empty {
return VersionCompatibility{
DetectedVersion: 0,
ExpectedVersion: expected,
Compatible: true,
PermitWrites: true,
Message: "cache database is uninitialized",
Remediation: "",
}, nil
}
return VersionCompatibility{
DetectedVersion: 0,
ExpectedVersion: expected,
Compatible: false,
PermitWrites: false,
Message: "cache database was created by a pre-schema-versioning binary (iteration 1 equivalent)",
Remediation: "confirm the selected cache path, move aside or delete only that cache file, then re-sync; for current-schema live data use 'gitcode-mcp cache reset --live --repo <repo>'; for live writes, rerun with '--cache-path <alternate-current-cache>' if another current-schema cache is available",
}, nil
}
detected, err := schemaVersion(ctx, db)
if err != nil {
return VersionCompatibility{}, err
}
if detected == 0 {
return VersionCompatibility{
DetectedVersion: 0,
ExpectedVersion: expected,
Compatible: false,
PermitWrites: false,
Message: "cache database contains an empty schema_version table (iteration 1 equivalent)",
Remediation: "confirm the selected cache path, move aside or delete only that cache file, then re-sync; for current-schema live data use 'gitcode-mcp cache reset --live --repo <repo>'; for live writes, rerun with '--cache-path <alternate-current-cache>' if another current-schema cache is available",
}, nil
}
if detected > expected {
return VersionCompatibility{
DetectedVersion: detected,
ExpectedVersion: expected,
Compatible: false,
PermitWrites: false,
Message: fmt.Sprintf("cache schema version %d is newer than supported version %d", detected, expected),
Remediation: "upgrade the gitcode-mcp binary to a version that supports this schema, or rerun with '--cache-path <alternate-current-cache>'",
}, nil
}
if detected < expected {
return VersionCompatibility{
DetectedVersion: detected,
ExpectedVersion: expected,
Compatible: true,
PermitWrites: false,
Message: fmt.Sprintf("cache schema version %d is older than expected version %d; writes are blocked until migration completes", detected, expected),
Remediation: fmt.Sprintf("run 'gitcode-mcp migrate-cache --confirm --cache-path <selected-cache>' to upgrade the schema from version %d to version %d, or rerun with '--cache-path <alternate-current-cache>'", detected, expected),
}, nil
}
return VersionCompatibility{
DetectedVersion: detected,
ExpectedVersion: expected,
Compatible: true,
PermitWrites: true,
Message: "cache schema is up to date",
Remediation: "",
}, nil
}
type migration struct {
version int
apply func(context.Context, *sql.Tx, bool) error
}
var migrations = []migration{
{version: 1, apply: applyInitialMigration},
{version: 2, apply: applyRepoScopedCacheMigration},
{version: 3, apply: applyChunkPolicyMigration},
{version: 4, apply: applyStoredSnapshotMigration},
{version: 5, apply: applySyncEventTimestampsMigration},
{version: 6, apply: applySyncEventZeroDeltaMigration},
{version: 7, apply: applyAuditIdempotencyMigration},
{version: 8, apply: applyCacheConfirmationsMigration},
{version: 9, apply: applyAuditConfirmationsMigration},
{version: 10, apply: applySourceOriginProvenanceMigration},
{version: 11, apply: applyRecordFixtureLiveProvenanceMigration},
{version: 12, apply: applyPRReviewCommentsMigration},
{version: 13, apply: applyPRReviewDiscussionPositionsMigration},
{version: 14, apply: applySyncFrontiersMigration},
{version: 15, apply: applyRAGEmbeddingSchemaMigration},
{version: 16, apply: applyIssueCommentSyncMigration},
{version: 17, apply: applyMaintenanceLifecycleMigration},
{version: 18, apply: applyRepositoryDocsSchemaMigration},
{version: 19, apply: applyRepositoryDocsIdentityMigration},
{version: 20, apply: applySyncCommitReceiptsMigration},
{version: 21, apply: applyMaintenanceLastSuccessMigration},
}
func runMigrations(ctx context.Context, db *sql.DB, ftsAvailable bool) error {
if _, err := db.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_version (version INTEGER NOT NULL)`); err != nil {
return err
}
version, err := schemaVersion(ctx, db)
if err != nil {
return err
}
if version > currentSchemaVersion {
return fmt.Errorf("cache: schema version %d is newer than supported version %d", version, currentSchemaVersion)
}
for _, m := range migrations {
if m.version <= version {
continue
}
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return err
}
if err = m.apply(ctx, tx, ftsAvailable); err != nil {
_ = tx.Rollback()
return err
}
if _, err = tx.ExecContext(ctx, `DELETE FROM schema_version`); err != nil {
_ = tx.Rollback()
return err
}
if _, err = tx.ExecContext(ctx, `INSERT INTO schema_version (version) VALUES (?)`, m.version); err != nil {
_ = tx.Rollback()
return err
}
if err = tx.Commit(); err != nil {
return err
}
version = m.version
}
return nil
}
func schemaVersion(ctx context.Context, db *sql.DB) (int, error) {
var count int
if err := db.QueryRowContext(ctx, `SELECT count(*) FROM schema_version`).Scan(&count); err != nil {
return 0, err
}
if count == 0 {
return 0, nil
}
if count > 1 {
return 0, fmt.Errorf("cache: schema_version must contain one row, found %d", count)
}
var version int
if err := db.QueryRowContext(ctx, `SELECT version FROM schema_version`).Scan(&version); err != nil {
return 0, err
}
return version, nil
}
func applyInitialMigration(ctx context.Context, tx *sql.Tx, ftsAvailable bool) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS repos (
repo_id TEXT PRIMARY KEY,
owner TEXT NOT NULL,
name TEXT NOT NULL,
api_base_url TEXT NOT NULL,
scopes TEXT NOT NULL,
display_name TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)`,
`CREATE TABLE IF NOT EXISTS repo_aliases (
alias TEXT PRIMARY KEY,
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
created_at TEXT NOT NULL
)`,
`CREATE INDEX IF NOT EXISTS idx_repo_aliases_repo ON repo_aliases(repo_id)`,
`CREATE TABLE IF NOT EXISTS sources (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
id TEXT NOT NULL,
kind TEXT NOT NULL,
path TEXT NOT NULL,
title TEXT NOT NULL,
body TEXT NOT NULL,
status TEXT NOT NULL,
labels TEXT NOT NULL,
content_hash TEXT NOT NULL,
provenance TEXT NOT NULL DEFAULT 'fixture' CHECK(provenance IN ('fixture', 'live')),
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY(repo_id, id)
)`,
`CREATE INDEX IF NOT EXISTS idx_sources_kind_status ON sources(repo_id, kind, status)`,
`CREATE TABLE IF NOT EXISTS identity_map (
repo_id TEXT NOT NULL,
source_id TEXT NOT NULL,
alias_type TEXT NOT NULL,
alias TEXT NOT NULL,
remote_type TEXT NOT NULL DEFAULT '',
remote_id TEXT NOT NULL DEFAULT '',
PRIMARY KEY(repo_id, alias_type, alias),
UNIQUE(repo_id, source_id, alias_type, alias),
FOREIGN KEY(repo_id, source_id) REFERENCES sources(repo_id, id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_identity_source ON identity_map(repo_id, source_id)`,
`CREATE INDEX IF NOT EXISTS idx_identity_remote ON identity_map(repo_id, remote_type, remote_id)`,
`CREATE TABLE IF NOT EXISTS links (
repo_id TEXT NOT NULL,
source_id TEXT NOT NULL,
target_id TEXT NOT NULL,
kind TEXT NOT NULL,
text TEXT NOT NULL,
PRIMARY KEY(repo_id, source_id, target_id, kind, text),
FOREIGN KEY(repo_id, source_id) REFERENCES sources(repo_id, id) ON DELETE CASCADE,
FOREIGN KEY(repo_id, target_id) REFERENCES sources(repo_id, id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_links_target ON links(repo_id, target_id)`,
`CREATE TABLE IF NOT EXISTS chunks (
repo_id TEXT NOT NULL,
id TEXT NOT NULL,
source_id TEXT NOT NULL,
record_id TEXT NOT NULL DEFAULT '',
snapshot_id TEXT NOT NULL DEFAULT '',
content_hash TEXT NOT NULL,
byte_start INTEGER NOT NULL,
byte_end INTEGER NOT NULL,
line_start INTEGER NOT NULL,
line_end INTEGER NOT NULL,
heading_path TEXT NOT NULL,
text TEXT NOT NULL,
normalized_text TEXT NOT NULL,
inherited_metadata TEXT NOT NULL,
outbound_links TEXT NOT NULL,
resolved_aliases TEXT NOT NULL,
embedding BLOB DEFAULT NULL,
policy TEXT NOT NULL DEFAULT 'heading',
PRIMARY KEY(repo_id, id),
UNIQUE(repo_id, source_id, content_hash, byte_start, policy, snapshot_id),
FOREIGN KEY(repo_id, source_id) REFERENCES sources(repo_id, id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_chunks_source ON chunks(repo_id, source_id)`,
`CREATE INDEX IF NOT EXISTS idx_chunks_query ON chunks(repo_id, source_id, record_id, snapshot_id, policy, byte_start, id)`,
`CREATE TABLE IF NOT EXISTS remote_revisions (
repo_id TEXT NOT NULL,
source_id TEXT NOT NULL,
remote_type TEXT NOT NULL,
remote_id TEXT NOT NULL,
remote_revision TEXT NOT NULL,
status TEXT NOT NULL,
last_fetched_at TEXT NOT NULL,
PRIMARY KEY(repo_id, source_id),
FOREIGN KEY(repo_id, source_id) REFERENCES sources(repo_id, id) ON DELETE CASCADE
)`,
`CREATE TABLE IF NOT EXISTS sync_events (
repo_id TEXT NOT NULL,
id TEXT NOT NULL,
source_id TEXT NOT NULL,
remote_type TEXT NOT NULL,
remote_id TEXT NOT NULL,
remote_revision TEXT NOT NULL,
status TEXT NOT NULL,
idempotency_key TEXT NOT NULL,
message TEXT NOT NULL,
created_at TEXT NOT NULL,
started_at TEXT NOT NULL DEFAULT '',
completed_at TEXT NOT NULL DEFAULT '',
zero_delta INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY(repo_id, id),
FOREIGN KEY(repo_id, source_id) REFERENCES sources(repo_id, id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_sync_events_source ON sync_events(repo_id, source_id)`,
`CREATE INDEX IF NOT EXISTS idx_sync_events_idempotency_key ON sync_events(repo_id, idempotency_key)`,
`CREATE TABLE IF NOT EXISTS sync_frontiers (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
remote_type TEXT NOT NULL,
ordering TEXT NOT NULL,
filter_key TEXT NOT NULL,
status TEXT NOT NULL,
high_updated_at TEXT NOT NULL,
high_remote_id TEXT NOT NULL,
high_number INTEGER NOT NULL DEFAULT 0,
stop_reason TEXT NOT NULL,
pages_listed INTEGER NOT NULL DEFAULT 0,
records_listed INTEGER NOT NULL DEFAULT 0,
updated_at TEXT NOT NULL,
PRIMARY KEY(repo_id, remote_type, ordering, filter_key)
)`,
`CREATE INDEX IF NOT EXISTS idx_sync_frontiers_repo ON sync_frontiers(repo_id, remote_type, status)`,
`CREATE TABLE IF NOT EXISTS conflicts (
repo_id TEXT NOT NULL,
id TEXT NOT NULL,
source_id TEXT NOT NULL,
kind TEXT NOT NULL,
local_payload TEXT NOT NULL,
remote_payload TEXT NOT NULL,
created_at TEXT NOT NULL,
PRIMARY KEY(repo_id, id),
FOREIGN KEY(repo_id, source_id) REFERENCES sources(repo_id, id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_conflicts_source ON conflicts(repo_id, source_id)`,
}
if ftsAvailable {
statements = append(statements, `CREATE VIRTUAL TABLE IF NOT EXISTS fts_index USING fts5(repo_id UNINDEXED, source_id UNINDEXED, path UNINDEXED, title, body)`)
}
for _, statement := range statements {
if _, err := tx.ExecContext(ctx, statement); err != nil {
return err
}
}
return nil
}
func applyRepoScopedCacheMigration(ctx context.Context, tx *sql.Tx, ftsAvailable bool) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS records (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
record_id TEXT NOT NULL,
record_type TEXT NOT NULL,
path TEXT NOT NULL,
title TEXT NOT NULL,
body TEXT NOT NULL,
status TEXT NOT NULL,
labels TEXT NOT NULL,
content_hash TEXT NOT NULL,
provenance TEXT NOT NULL CHECK(provenance IN ('remote', 'projection', 'bridge')),
remote_type TEXT NOT NULL DEFAULT '',
remote_id TEXT NOT NULL DEFAULT '',
remote_revision TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY(repo_id, record_id)
)`,
`CREATE INDEX IF NOT EXISTS idx_records_type_status ON records(repo_id, record_type, status)`,
`CREATE INDEX IF NOT EXISTS idx_records_remote ON records(repo_id, remote_type, remote_id)`,
`CREATE UNIQUE INDEX IF NOT EXISTS idx_records_remote_unique ON records(repo_id, remote_type, remote_id) WHERE remote_type <> '' AND remote_id <> ''`,
`CREATE TABLE IF NOT EXISTS record_comments (
repo_id TEXT NOT NULL,
record_id TEXT NOT NULL,
comment_id TEXT NOT NULL,
author TEXT NOT NULL,
body TEXT NOT NULL,
content_hash TEXT NOT NULL,
remote_revision TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY(repo_id, record_id, comment_id),
FOREIGN KEY(repo_id, record_id) REFERENCES records(repo_id, record_id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_record_comments_record ON record_comments(repo_id, record_id)`,
`CREATE TABLE IF NOT EXISTS audit_trail (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
id TEXT NOT NULL,
operation TEXT NOT NULL,
record_id TEXT NOT NULL DEFAULT '',
remote_type TEXT NOT NULL DEFAULT '',
remote_id TEXT NOT NULL DEFAULT '',
idempotency_key TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL,
message TEXT NOT NULL DEFAULT '',
payload_hash TEXT NOT NULL DEFAULT '',
command TEXT NOT NULL DEFAULT '',
mode TEXT NOT NULL DEFAULT '',
request_metadata TEXT NOT NULL DEFAULT '{}',
created_at TEXT NOT NULL,
PRIMARY KEY(repo_id, id)
)`,
`CREATE INDEX IF NOT EXISTS idx_audit_trail_record ON audit_trail(repo_id, record_id)`,
`CREATE INDEX IF NOT EXISTS idx_audit_trail_idempotency ON audit_trail(repo_id, idempotency_key)`,
`CREATE UNIQUE INDEX IF NOT EXISTS idx_audit_trail_idempotency_unique ON audit_trail(repo_id, idempotency_key) WHERE idempotency_key <> ''`,
`CREATE TABLE IF NOT EXISTS snapshots (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
snapshot_id TEXT NOT NULL,
format TEXT NOT NULL,
content_hash TEXT NOT NULL,
record_count INTEGER NOT NULL,
created_at TEXT NOT NULL,
schema_version TEXT NOT NULL DEFAULT 'gitcode-mcp.snapshot.v1',
manifest_hash TEXT NOT NULL DEFAULT '',
chunk_set_hash TEXT NOT NULL DEFAULT '',
chunk_count INTEGER NOT NULL DEFAULT 0,
manifest_json TEXT NOT NULL DEFAULT '{}',
warnings_json TEXT NOT NULL DEFAULT '[]',
metadata TEXT NOT NULL DEFAULT '{}',
PRIMARY KEY(repo_id, snapshot_id)
)`,
`CREATE TABLE IF NOT EXISTS snapshot_chunks (
repo_id TEXT NOT NULL,
snapshot_id TEXT NOT NULL,
chunk_id TEXT NOT NULL,
source_type TEXT NOT NULL DEFAULT '',
source_id TEXT NOT NULL DEFAULT '',
record_id TEXT NOT NULL,
source_content_hash TEXT NOT NULL DEFAULT '',
source_revision_hash TEXT NOT NULL DEFAULT '',
index_build_id TEXT NOT NULL DEFAULT '',
chunk_content_hash TEXT NOT NULL DEFAULT '',
byte_start INTEGER NOT NULL,
byte_end INTEGER NOT NULL,
line_start INTEGER NOT NULL,
line_end INTEGER NOT NULL,
heading_path TEXT NOT NULL DEFAULT '[]',
ordinal INTEGER NOT NULL DEFAULT 0,
text TEXT NOT NULL DEFAULT '',
metadata_json TEXT NOT NULL DEFAULT '{}',
outbound_links_json TEXT NOT NULL DEFAULT '[]',
resolved_aliases_json TEXT NOT NULL DEFAULT '{}',
citation TEXT NOT NULL DEFAULT '',
content_hash TEXT NOT NULL DEFAULT '',
PRIMARY KEY(repo_id, snapshot_id, chunk_id),
FOREIGN KEY(repo_id, snapshot_id) REFERENCES snapshots(repo_id, snapshot_id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_snapshot_chunks_record ON snapshot_chunks(repo_id, record_id)`,
`CREATE UNIQUE INDEX IF NOT EXISTS idx_remote_revisions_remote_unique ON remote_revisions(repo_id, remote_type, remote_id)`,
`CREATE UNIQUE INDEX IF NOT EXISTS idx_sync_events_idempotency_unique ON sync_events(repo_id, idempotency_key)`,
}
for _, statement := range statements {
if _, err := tx.ExecContext(ctx, statement); err != nil {
return err
}
}
return nil
}
func applyChunkPolicyMigration(ctx context.Context, tx *sql.Tx, ftsAvailable bool) error {
columns, err := tableColumns(ctx, tx, "chunks")
if err != nil {
return err
}
addColumns := map[string]string{
"record_id": `ALTER TABLE chunks ADD COLUMN record_id TEXT NOT NULL DEFAULT ''`,
"snapshot_id": `ALTER TABLE chunks ADD COLUMN snapshot_id TEXT NOT NULL DEFAULT ''`,
"policy": `ALTER TABLE chunks ADD COLUMN policy TEXT NOT NULL DEFAULT 'heading'`,
}
for column, statement := range addColumns {
if columns[column] {
continue
}
if _, err := tx.ExecContext(ctx, statement); err != nil {
return err
}
}
_, err = tx.ExecContext(ctx, `CREATE INDEX IF NOT EXISTS idx_chunks_query ON chunks(repo_id, source_id, record_id, snapshot_id, policy, byte_start, id)`)
return err
}
func applyStoredSnapshotMigration(ctx context.Context, tx *sql.Tx, ftsAvailable bool) error {
snapshotColumns, err := tableColumns(ctx, tx, "snapshots")
if err != nil {
return err
}
snapshotAdds := map[string]string{
"schema_version": `ALTER TABLE snapshots ADD COLUMN schema_version TEXT NOT NULL DEFAULT 'gitcode-mcp.snapshot.v1'`,
"manifest_hash": `ALTER TABLE snapshots ADD COLUMN manifest_hash TEXT NOT NULL DEFAULT ''`,
"chunk_set_hash": `ALTER TABLE snapshots ADD COLUMN chunk_set_hash TEXT NOT NULL DEFAULT ''`,
"chunk_count": `ALTER TABLE snapshots ADD COLUMN chunk_count INTEGER NOT NULL DEFAULT 0`,
"manifest_json": `ALTER TABLE snapshots ADD COLUMN manifest_json TEXT NOT NULL DEFAULT '{}'`,
"warnings_json": `ALTER TABLE snapshots ADD COLUMN warnings_json TEXT NOT NULL DEFAULT '[]'`,
}
for column, statement := range snapshotAdds {
if !snapshotColumns[column] {
if _, err := tx.ExecContext(ctx, statement); err != nil {
return err
}
}
}
chunkColumns, err := tableColumns(ctx, tx, "snapshot_chunks")
if err != nil {
return err
}
chunkAdds := map[string]string{
"source_type": `ALTER TABLE snapshot_chunks ADD COLUMN source_type TEXT NOT NULL DEFAULT ''`,
"source_id": `ALTER TABLE snapshot_chunks ADD COLUMN source_id TEXT NOT NULL DEFAULT ''`,
"source_content_hash": `ALTER TABLE snapshot_chunks ADD COLUMN source_content_hash TEXT NOT NULL DEFAULT ''`,
"source_revision_hash": `ALTER TABLE snapshot_chunks ADD COLUMN source_revision_hash TEXT NOT NULL DEFAULT ''`,
"index_build_id": `ALTER TABLE snapshot_chunks ADD COLUMN index_build_id TEXT NOT NULL DEFAULT ''`,
"chunk_content_hash": `ALTER TABLE snapshot_chunks ADD COLUMN chunk_content_hash TEXT NOT NULL DEFAULT ''`,
"heading_path": `ALTER TABLE snapshot_chunks ADD COLUMN heading_path TEXT NOT NULL DEFAULT '[]'`,
"ordinal": `ALTER TABLE snapshot_chunks ADD COLUMN ordinal INTEGER NOT NULL DEFAULT 0`,
"text": `ALTER TABLE snapshot_chunks ADD COLUMN text TEXT NOT NULL DEFAULT ''`,
"metadata_json": `ALTER TABLE snapshot_chunks ADD COLUMN metadata_json TEXT NOT NULL DEFAULT '{}'`,
"outbound_links_json": `ALTER TABLE snapshot_chunks ADD COLUMN outbound_links_json TEXT NOT NULL DEFAULT '[]'`,
"resolved_aliases_json": `ALTER TABLE snapshot_chunks ADD COLUMN resolved_aliases_json TEXT NOT NULL DEFAULT '{}'`,
}
for column, statement := range chunkAdds {
if !chunkColumns[column] {
if _, err := tx.ExecContext(ctx, statement); err != nil {
return err
}
}
}
_, err = tx.ExecContext(ctx, `CREATE INDEX IF NOT EXISTS idx_snapshot_chunks_order ON snapshot_chunks(repo_id, snapshot_id, source_type, source_id, record_id, ordinal, chunk_id)`)
return err
}
func applySyncEventTimestampsMigration(ctx context.Context, tx *sql.Tx, ftsAvailable bool) error {
columns, err := tableColumns(ctx, tx, "sync_events")
if err != nil {
return err
}
addColumns := map[string]string{
"started_at": `ALTER TABLE sync_events ADD COLUMN started_at TEXT NOT NULL DEFAULT ''`,
"completed_at": `ALTER TABLE sync_events ADD COLUMN completed_at TEXT NOT NULL DEFAULT ''`,
}
for column, statement := range addColumns {
if columns[column] {
continue
}
if _, err := tx.ExecContext(ctx, statement); err != nil {
return err
}
}
return nil
}
func applySyncEventZeroDeltaMigration(ctx context.Context, tx *sql.Tx, ftsAvailable bool) error {
columns, err := tableColumns(ctx, tx, "sync_events")
if err != nil {
return err
}
if columns["zero_delta"] {
return nil
}
_, err = tx.ExecContext(ctx, `ALTER TABLE sync_events ADD COLUMN zero_delta INTEGER NOT NULL DEFAULT 0`)
return err
}
func applyAuditIdempotencyMigration(ctx context.Context, tx *sql.Tx, ftsAvailable bool) error {
_, err := tx.ExecContext(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_audit_trail_idempotency_unique ON audit_trail(repo_id, idempotency_key) WHERE idempotency_key <> ''`)
return err
}
func applyCacheConfirmationsMigration(ctx context.Context, tx *sql.Tx, ftsAvailable bool) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS cache_confirmations (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
id TEXT NOT NULL,
command TEXT NOT NULL,
record_id TEXT NOT NULL,
record_type TEXT NOT NULL DEFAULT '',
remote_type TEXT NOT NULL,
remote_id TEXT NOT NULL,
idempotency_key TEXT NOT NULL,
status TEXT NOT NULL,
source_fingerprint TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
PRIMARY KEY(repo_id, id),
UNIQUE(repo_id, idempotency_key),
FOREIGN KEY(repo_id, record_id) REFERENCES records(repo_id, record_id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_cache_confirmations_record ON cache_confirmations(repo_id, record_id)`,
`CREATE INDEX IF NOT EXISTS idx_cache_confirmations_remote ON cache_confirmations(repo_id, remote_type, remote_id)`,
}
for _, statement := range statements {
if _, err := tx.ExecContext(ctx, statement); err != nil {
return err
}
}
return nil
}
func applyAuditConfirmationsMigration(ctx context.Context, tx *sql.Tx, ftsAvailable bool) error {
columns, err := tableColumns(ctx, tx, "audit_trail")
if err != nil {
return err
}
addColumns := map[string]string{
"command": `ALTER TABLE audit_trail ADD COLUMN command TEXT NOT NULL DEFAULT ''`,
"mode": `ALTER TABLE audit_trail ADD COLUMN mode TEXT NOT NULL DEFAULT ''`,
"request_metadata": `ALTER TABLE audit_trail ADD COLUMN request_metadata TEXT NOT NULL DEFAULT '{}'`,
}
for column, statement := range addColumns {
if columns[column] {
continue
}
if _, err := tx.ExecContext(ctx, statement); err != nil {
return err
}
}
return nil
}
func applySourceOriginProvenanceMigration(ctx context.Context, tx *sql.Tx, ftsAvailable bool) error {
columns, err := tableColumns(ctx, tx, "sources")
if err != nil {
return err
}
if columns["provenance"] {
return nil
}
_, err = tx.ExecContext(ctx, `ALTER TABLE sources ADD COLUMN provenance TEXT NOT NULL DEFAULT 'fixture' CHECK(provenance IN ('fixture', 'live'))`)
return err
}
func applyRecordFixtureLiveProvenanceMigration(ctx context.Context, tx *sql.Tx, ftsAvailable bool) error {
if _, err := tx.ExecContext(ctx, `PRAGMA foreign_keys = OFF`); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS records_new (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
record_id TEXT NOT NULL,
record_type TEXT NOT NULL,
path TEXT NOT NULL,
title TEXT NOT NULL,
body TEXT NOT NULL,
status TEXT NOT NULL,
labels TEXT NOT NULL,
content_hash TEXT NOT NULL,
provenance TEXT NOT NULL CHECK(provenance IN ('remote', 'projection', 'bridge', 'fixture', 'live')),
remote_type TEXT NOT NULL DEFAULT '',
remote_id TEXT NOT NULL DEFAULT '',
remote_revision TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY(repo_id, record_id)
)`); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `INSERT INTO records_new SELECT * FROM records`); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `DROP TABLE records`); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `ALTER TABLE records_new RENAME TO records`); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `PRAGMA foreign_keys = ON`); err != nil {
return err
}
for _, stmt := range []string{
`CREATE INDEX IF NOT EXISTS idx_records_type_status ON records(repo_id, record_type, status)`,
`CREATE INDEX IF NOT EXISTS idx_records_remote ON records(repo_id, remote_type, remote_id)`,
`CREATE UNIQUE INDEX IF NOT EXISTS idx_records_remote_unique ON records(repo_id, remote_type, remote_id) WHERE remote_type <> '' AND remote_id <> ''`,
} {
if _, err := tx.ExecContext(ctx, stmt); err != nil {
return err
}
}
return nil
}
func applyPRReviewCommentsMigration(ctx context.Context, tx *sql.Tx, ftsAvailable bool) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS pr_review_comments (
repo_id TEXT NOT NULL,
source_id TEXT NOT NULL,
pr_number INTEGER NOT NULL,
comment_id TEXT NOT NULL,
discussion_id TEXT NOT NULL DEFAULT '',
review_kind TEXT NOT NULL DEFAULT '',
author TEXT NOT NULL DEFAULT '',
path TEXT NOT NULL DEFAULT '',
line INTEGER NOT NULL DEFAULT 0,
start_line INTEGER NOT NULL DEFAULT 0,
end_line INTEGER NOT NULL DEFAULT 0,
position INTEGER NOT NULL DEFAULT 0,
original_position INTEGER NOT NULL DEFAULT 0,
resolved TEXT NOT NULL DEFAULT '',
resolvable TEXT NOT NULL DEFAULT '',
parent_id TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY(repo_id, source_id),
FOREIGN KEY(repo_id, source_id) REFERENCES sources(repo_id, id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_pr_review_comments_pr ON pr_review_comments(repo_id, pr_number)`,
`CREATE INDEX IF NOT EXISTS idx_pr_review_comments_discussion ON pr_review_comments(repo_id, pr_number, discussion_id)`,
}
for _, stmt := range statements {
if _, err := tx.ExecContext(ctx, stmt); err != nil {
return err
}
}
return nil
}
func applyPRReviewDiscussionPositionsMigration(ctx context.Context, tx *sql.Tx, ftsAvailable bool) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS pr_review_discussions (
repo_id TEXT NOT NULL,
pr_number INTEGER NOT NULL,
discussion_id TEXT NOT NULL,
kind TEXT NOT NULL DEFAULT '',
resolved TEXT NOT NULL DEFAULT '',
resolvable TEXT NOT NULL DEFAULT '',
first_comment_id TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY(repo_id, pr_number, discussion_id),
FOREIGN KEY(repo_id) REFERENCES repos(repo_id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_pr_review_discussions_pr ON pr_review_discussions(repo_id, pr_number)`,
`CREATE TABLE IF NOT EXISTS pr_review_positions (
repo_id TEXT NOT NULL,
pr_number INTEGER NOT NULL,
comment_id TEXT NOT NULL,
position_kind TEXT NOT NULL DEFAULT 'current',
discussion_id TEXT NOT NULL DEFAULT '',
position_type TEXT NOT NULL DEFAULT '',
base_sha TEXT NOT NULL DEFAULT '',
start_sha TEXT NOT NULL DEFAULT '',
head_sha TEXT NOT NULL DEFAULT '',
old_path TEXT NOT NULL DEFAULT '',
new_path TEXT NOT NULL DEFAULT '',
old_line INTEGER NOT NULL DEFAULT 0,
new_line INTEGER NOT NULL DEFAULT 0,
start_old_line INTEGER NOT NULL DEFAULT 0,
start_new_line INTEGER NOT NULL DEFAULT 0,
line_code TEXT NOT NULL DEFAULT '',
start_line_code TEXT NOT NULL DEFAULT '',
patchset_iid INTEGER NOT NULL DEFAULT 0,
diff_id INTEGER NOT NULL DEFAULT 0,
version_sha TEXT NOT NULL DEFAULT '',
side TEXT NOT NULL DEFAULT '',
is_outdated TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY(repo_id, pr_number, comment_id, position_kind),
FOREIGN KEY(repo_id) REFERENCES repos(repo_id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_pr_review_positions_discussion ON pr_review_positions(repo_id, pr_number, discussion_id)`,
`CREATE INDEX IF NOT EXISTS idx_pr_review_positions_new_line ON pr_review_positions(repo_id, pr_number, new_path, new_line)`,
`CREATE INDEX IF NOT EXISTS idx_pr_review_positions_old_line ON pr_review_positions(repo_id, pr_number, old_path, old_line)`,
}
for _, stmt := range statements {
if _, err := tx.ExecContext(ctx, stmt); err != nil {
return err
}
}
return nil
}
func applySyncFrontiersMigration(ctx context.Context, tx *sql.Tx, _ bool) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS sync_frontiers (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
remote_type TEXT NOT NULL,
ordering TEXT NOT NULL,
filter_key TEXT NOT NULL,
status TEXT NOT NULL,
high_updated_at TEXT NOT NULL,
high_remote_id TEXT NOT NULL,
high_number INTEGER NOT NULL DEFAULT 0,
stop_reason TEXT NOT NULL,
pages_listed INTEGER NOT NULL DEFAULT 0,
records_listed INTEGER NOT NULL DEFAULT 0,
updated_at TEXT NOT NULL,
PRIMARY KEY(repo_id, remote_type, ordering, filter_key)
)`,
`CREATE INDEX IF NOT EXISTS idx_sync_frontiers_repo ON sync_frontiers(repo_id, remote_type, status)`,
}
for _, stmt := range statements {
if _, err := tx.ExecContext(ctx, stmt); err != nil {
return err
}
}
return nil
}
func applyIssueCommentSyncMigration(ctx context.Context, tx *sql.Tx, _ bool) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS issue_comment_sync (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
source_id TEXT NOT NULL,
issue_number INTEGER NOT NULL,
remote_id TEXT NOT NULL,
provider_id TEXT NOT NULL DEFAULT '',
remote_revision TEXT NOT NULL,
expected_count INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL CHECK(status IN ('pending', 'deferred', 'complete')),
attempts INTEGER NOT NULL DEFAULT 0,
last_error_class TEXT NOT NULL DEFAULT '',
retry_after TEXT NOT NULL DEFAULT '',
last_attempt_at TEXT NOT NULL DEFAULT '',
updated_at TEXT NOT NULL,
PRIMARY KEY(repo_id, source_id),
FOREIGN KEY(repo_id, source_id) REFERENCES sources(repo_id, id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_issue_comment_sync_queue ON issue_comment_sync(repo_id, status, updated_at, issue_number)`,
}
for _, stmt := range statements {
if _, err := tx.ExecContext(ctx, stmt); err != nil {
return err
}
}
return nil
}
func applyRAGEmbeddingSchemaMigration(ctx context.Context, tx *sql.Tx, _ bool) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS embedding_namespaces (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
namespace_id TEXT NOT NULL,
profile_id TEXT NOT NULL,
provider_id TEXT NOT NULL,
provider_type TEXT NOT NULL,
model_id TEXT NOT NULL,
model_revision TEXT NOT NULL DEFAULT '',
dimensions INTEGER NOT NULL,
dtype TEXT NOT NULL,
normalization TEXT NOT NULL,
document_instruction_id TEXT NOT NULL DEFAULT '',
query_instruction_id TEXT NOT NULL DEFAULT '',
chunk_policy_id TEXT NOT NULL,
language_policy_id TEXT NOT NULL,
config_hash TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY(repo_id, namespace_id)
)`,
`CREATE UNIQUE INDEX IF NOT EXISTS idx_embedding_namespaces_identity ON embedding_namespaces(repo_id, provider_id, provider_type, model_id, model_revision, dimensions, dtype, normalization, document_instruction_id, query_instruction_id, chunk_policy_id, language_policy_id, config_hash)`,
`CREATE INDEX IF NOT EXISTS idx_embedding_namespaces_profile ON embedding_namespaces(repo_id, profile_id)`,
`CREATE TABLE IF NOT EXISTS chunk_embeddings (
repo_id TEXT NOT NULL,
namespace_id TEXT NOT NULL,
chunk_id TEXT NOT NULL,
source_id TEXT NOT NULL DEFAULT '',
record_id TEXT NOT NULL DEFAULT '',
snapshot_id TEXT NOT NULL DEFAULT '',
chunk_content_hash TEXT NOT NULL,
vector BLOB NOT NULL,
dimensions INTEGER NOT NULL,
dtype TEXT NOT NULL,
vector_hash TEXT NOT NULL,
embedded_at TEXT NOT NULL,
PRIMARY KEY(repo_id, namespace_id, chunk_id),
FOREIGN KEY(repo_id, namespace_id) REFERENCES embedding_namespaces(repo_id, namespace_id) ON DELETE CASCADE,
FOREIGN KEY(repo_id, chunk_id) REFERENCES chunks(repo_id, id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_chunk_embeddings_chunk ON chunk_embeddings(repo_id, chunk_id)`,
`CREATE INDEX IF NOT EXISTS idx_chunk_embeddings_coverage ON chunk_embeddings(repo_id, namespace_id, source_id, record_id, snapshot_id)`,
`CREATE TABLE IF NOT EXISTS rag_index_runs (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
run_id TEXT NOT NULL,
namespace_id TEXT NOT NULL,
profile_id TEXT NOT NULL,
status TEXT NOT NULL,
total_chunks INTEGER NOT NULL DEFAULT 0,
embedded_chunks INTEGER NOT NULL DEFAULT 0,
skipped_chunks INTEGER NOT NULL DEFAULT 0,
failed_chunks INTEGER NOT NULL DEFAULT 0,
started_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
completed_at TEXT NOT NULL DEFAULT '',
error_class TEXT NOT NULL DEFAULT '',
message TEXT NOT NULL DEFAULT '',
metadata_json TEXT NOT NULL DEFAULT '{}',
PRIMARY KEY(repo_id, run_id),
FOREIGN KEY(repo_id, namespace_id) REFERENCES embedding_namespaces(repo_id, namespace_id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_rag_index_runs_namespace ON rag_index_runs(repo_id, namespace_id, started_at)`,
`CREATE INDEX IF NOT EXISTS idx_rag_index_runs_status ON rag_index_runs(repo_id, status, updated_at)`,
}
for _, stmt := range statements {
if _, err := tx.ExecContext(ctx, stmt); err != nil {
return err
}
}
return nil
}
func applyMaintenanceLifecycleMigration(ctx context.Context, tx *sql.Tx, _ bool) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS cache_identity (
identity_key INTEGER PRIMARY KEY CHECK(identity_key = 1),
cache_uuid TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL
)`,
`INSERT OR IGNORE INTO cache_identity (identity_key, cache_uuid, created_at)
VALUES (1, 'cache-' || lower(hex(randomblob(16))), strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))`,
`CREATE TABLE IF NOT EXISTS repo_content_state (
repo_id TEXT PRIMARY KEY REFERENCES repos(repo_id) ON DELETE CASCADE,
content_generation INTEGER NOT NULL DEFAULT 0,
content_changed_at TEXT NOT NULL DEFAULT '',
last_projection_id TEXT NOT NULL DEFAULT ''
)`,
`INSERT OR IGNORE INTO repo_content_state (repo_id, content_generation, content_changed_at, last_projection_id)
SELECT r.repo_id, 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), 'migration-v17'
FROM repos r
WHERE EXISTS (SELECT 1 FROM sources s WHERE s.repo_id = r.repo_id)
OR EXISTS (SELECT 1 FROM chunks c WHERE c.repo_id = r.repo_id)`,
`CREATE TABLE IF NOT EXISTS rag_coverage_state (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
namespace_id TEXT NOT NULL,
covered_generation INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY(repo_id, namespace_id),
FOREIGN KEY(repo_id, namespace_id) REFERENCES embedding_namespaces(repo_id, namespace_id) ON DELETE CASCADE
)`,
`CREATE TABLE IF NOT EXISTS maintenance_frontiers (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
remote_type TEXT NOT NULL,
ordering TEXT NOT NULL,
filter_key TEXT NOT NULL,
lane TEXT NOT NULL CHECK(lane IN ('head', 'tail', 'secondary')),
status TEXT NOT NULL,
high_updated_at TEXT NOT NULL DEFAULT '',
high_remote_id TEXT NOT NULL DEFAULT '',
high_number INTEGER NOT NULL DEFAULT 0,
stop_reason TEXT NOT NULL DEFAULT '',
pages_listed INTEGER NOT NULL DEFAULT 0,
records_listed INTEGER NOT NULL DEFAULT 0,
checkpoint TEXT NOT NULL DEFAULT '',
last_error_class TEXT NOT NULL DEFAULT '',
updated_at TEXT NOT NULL,
PRIMARY KEY(repo_id, remote_type, ordering, filter_key, lane)
)`,
`CREATE INDEX IF NOT EXISTS idx_maintenance_frontiers_state ON maintenance_frontiers(repo_id, lane, status, updated_at)`,
`CREATE TRIGGER IF NOT EXISTS trg_sources_content_insert AFTER INSERT ON sources
BEGIN
INSERT INTO repo_content_state (repo_id, content_generation, content_changed_at)
VALUES (NEW.repo_id, 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
ON CONFLICT(repo_id) DO UPDATE SET content_generation = content_generation + 1, content_changed_at = excluded.content_changed_at;
END`,
`CREATE TRIGGER IF NOT EXISTS trg_sources_content_update AFTER UPDATE OF content_hash ON sources
WHEN OLD.content_hash <> NEW.content_hash
BEGIN
INSERT INTO repo_content_state (repo_id, content_generation, content_changed_at)
VALUES (NEW.repo_id, 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
ON CONFLICT(repo_id) DO UPDATE SET content_generation = content_generation + 1, content_changed_at = excluded.content_changed_at;
END`,
`CREATE TRIGGER IF NOT EXISTS trg_sources_content_delete AFTER DELETE ON sources
BEGIN
INSERT INTO repo_content_state (repo_id, content_generation, content_changed_at)
VALUES (OLD.repo_id, 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
ON CONFLICT(repo_id) DO UPDATE SET content_generation = content_generation + 1, content_changed_at = excluded.content_changed_at;
END`,
`CREATE TRIGGER IF NOT EXISTS trg_chunks_content_insert AFTER INSERT ON chunks
BEGIN
INSERT INTO repo_content_state (repo_id, content_generation, content_changed_at)
VALUES (NEW.repo_id, 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
ON CONFLICT(repo_id) DO UPDATE SET content_generation = content_generation + 1, content_changed_at = excluded.content_changed_at;
END`,
`CREATE TRIGGER IF NOT EXISTS trg_chunks_content_update AFTER UPDATE OF content_hash ON chunks
WHEN OLD.content_hash <> NEW.content_hash
BEGIN
INSERT INTO repo_content_state (repo_id, content_generation, content_changed_at)
VALUES (NEW.repo_id, 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
ON CONFLICT(repo_id) DO UPDATE SET content_generation = content_generation + 1, content_changed_at = excluded.content_changed_at;
END`,
`CREATE TRIGGER IF NOT EXISTS trg_chunks_content_delete AFTER DELETE ON chunks
BEGIN
INSERT INTO repo_content_state (repo_id, content_generation, content_changed_at)
VALUES (OLD.repo_id, 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
ON CONFLICT(repo_id) DO UPDATE SET content_generation = content_generation + 1, content_changed_at = excluded.content_changed_at;
END`,
}
for _, stmt := range statements {
if _, err := tx.ExecContext(ctx, stmt); err != nil {
return err
}
}
return nil
}
func applyRepositoryDocsSchemaMigration(ctx context.Context, tx *sql.Tx, _ bool) error {
statements := []string{
`CREATE TABLE IF NOT EXISTS repo_doc_revision_sets (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
revision_set_id TEXT NOT NULL,
git_store_ref TEXT NOT NULL,
worktree_ref TEXT NOT NULL DEFAULT '',
object_format TEXT NOT NULL,
commit_oid TEXT NOT NULL,
requested_revision TEXT NOT NULL DEFAULT '',
policy_hash TEXT NOT NULL,
policy_source TEXT NOT NULL,
config_digest TEXT NOT NULL DEFAULT '',
overlay_digest TEXT NOT NULL DEFAULT '',
chunk_policy_id TEXT NOT NULL,
namespace_id TEXT NOT NULL DEFAULT '',
state TEXT NOT NULL,
eligible_files INTEGER NOT NULL DEFAULT 0,
eligible_chunks INTEGER NOT NULL DEFAULT 0,
embedded_chunks INTEGER NOT NULL DEFAULT 0,
reused_chunks INTEGER NOT NULL DEFAULT 0,
failed_chunks INTEGER NOT NULL DEFAULT 0,
excluded_files INTEGER NOT NULL DEFAULT 0,
missing_objects INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
completed_at TEXT NOT NULL DEFAULT '',
last_error_class TEXT NOT NULL DEFAULT '',
PRIMARY KEY(repo_id, revision_set_id)
)`,
`CREATE INDEX IF NOT EXISTS idx_repo_doc_revision_sets_lookup ON repo_doc_revision_sets(repo_id, git_store_ref, commit_oid, policy_hash, overlay_digest, chunk_policy_id, state, updated_at)`,
`CREATE INDEX IF NOT EXISTS idx_repo_doc_revision_sets_retention ON repo_doc_revision_sets(repo_id, state, completed_at, updated_at)`,
`CREATE TABLE IF NOT EXISTS repo_doc_chunks (
repo_id TEXT NOT NULL REFERENCES repos(repo_id) ON DELETE CASCADE,
chunk_id TEXT NOT NULL,
object_format TEXT NOT NULL,
blob_oid TEXT NOT NULL DEFAULT '',
worktree_ref TEXT NOT NULL DEFAULT '',
content_digest TEXT NOT NULL,
byte_start INTEGER NOT NULL,
byte_end INTEGER NOT NULL,
line_start INTEGER NOT NULL,
line_end INTEGER NOT NULL,
raw_slice_digest TEXT NOT NULL,
embedding_input_digest TEXT NOT NULL,
chunk_policy_id TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY(repo_id, chunk_id)
)`,
`CREATE INDEX IF NOT EXISTS idx_repo_doc_chunks_content ON repo_doc_chunks(repo_id, object_format, blob_oid, content_digest, chunk_policy_id)`,
`CREATE TABLE IF NOT EXISTS repo_doc_membership (
repo_id TEXT NOT NULL,
revision_set_id TEXT NOT NULL,
path TEXT NOT NULL,
chunk_id TEXT NOT NULL,
authority TEXT NOT NULL,
ordinal INTEGER NOT NULL,
blob_oid TEXT NOT NULL DEFAULT '',
worktree_ref TEXT NOT NULL DEFAULT '',
content_digest TEXT NOT NULL,
PRIMARY KEY(repo_id, revision_set_id, path, chunk_id),
FOREIGN KEY(repo_id, revision_set_id) REFERENCES repo_doc_revision_sets(repo_id, revision_set_id) ON DELETE CASCADE,
FOREIGN KEY(repo_id, chunk_id) REFERENCES repo_doc_chunks(repo_id, chunk_id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_repo_doc_membership_chunk ON repo_doc_membership(repo_id, chunk_id, revision_set_id)`,
`CREATE INDEX IF NOT EXISTS idx_repo_doc_membership_path ON repo_doc_membership(repo_id, revision_set_id, path, ordinal)`,
`CREATE TABLE IF NOT EXISTS repo_doc_vectors (
repo_id TEXT NOT NULL,
namespace_id TEXT NOT NULL,
chunk_id TEXT NOT NULL,
vector BLOB NOT NULL,
dimensions INTEGER NOT NULL,
dtype TEXT NOT NULL,
vector_hash TEXT NOT NULL,
embedded_at TEXT NOT NULL,
PRIMARY KEY(repo_id, namespace_id, chunk_id),
FOREIGN KEY(repo_id, namespace_id) REFERENCES embedding_namespaces(repo_id, namespace_id) ON DELETE CASCADE,
FOREIGN KEY(repo_id, chunk_id) REFERENCES repo_doc_chunks(repo_id, chunk_id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_repo_doc_vectors_chunk ON repo_doc_vectors(repo_id, chunk_id, namespace_id)`,
}
for _, stmt := range statements {
if _, err := tx.ExecContext(ctx, stmt); err != nil {
return err
}
}
return nil
}
func applyRepositoryDocsIdentityMigration(ctx context.Context, tx *sql.Tx, _ bool) error {
columns, err := tableColumns(ctx, tx, "repo_doc_revision_sets")
if err != nil {
return err
}
for _, column := range []struct {
name string
sql string
}{
{"source_registration_id", `ALTER TABLE repo_doc_revision_sets ADD COLUMN source_registration_id TEXT NOT NULL DEFAULT ''`},
{"source_registration_generation", `ALTER TABLE repo_doc_revision_sets ADD COLUMN source_registration_generation INTEGER NOT NULL DEFAULT 0`},
{"processing_policy_id", `ALTER TABLE repo_doc_revision_sets ADD COLUMN processing_policy_id TEXT NOT NULL DEFAULT ''`},
} {
if columns[column.name] {
continue
}
if _, err := tx.ExecContext(ctx, column.sql); err != nil {
return err
}
}
membershipColumns, err := tableColumns(ctx, tx, "repo_doc_membership")
if err != nil {
return err
}
if !membershipColumns["worktree_ref"] {
if _, err := tx.ExecContext(ctx, `ALTER TABLE repo_doc_membership ADD COLUMN worktree_ref TEXT NOT NULL DEFAULT ''`); err != nil {
return err
}
}
for _, stmt := range []string{
`DROP INDEX IF EXISTS idx_repo_doc_revision_sets_lookup`,
`CREATE INDEX idx_repo_doc_revision_sets_lookup ON repo_doc_revision_sets(repo_id, source_registration_id, source_registration_generation, git_store_ref, commit_oid, policy_hash, overlay_digest, processing_policy_id, namespace_id, state, updated_at)`,
`CREATE TABLE IF NOT EXISTS repo_doc_exclusions (
repo_id TEXT NOT NULL,
revision_set_id TEXT NOT NULL,
path TEXT NOT NULL,
authority TEXT NOT NULL,
blob_oid TEXT NOT NULL DEFAULT '',
reason_code TEXT NOT NULL,
PRIMARY KEY(repo_id, revision_set_id, path, reason_code),
FOREIGN KEY(repo_id, revision_set_id) REFERENCES repo_doc_revision_sets(repo_id, revision_set_id) ON DELETE CASCADE
)`,
`CREATE INDEX IF NOT EXISTS idx_repo_doc_exclusions_reason ON repo_doc_exclusions(repo_id, revision_set_id, reason_code, path)`,
} {
if _, err := tx.ExecContext(ctx, stmt); err != nil {
return err
}
}
return nil
}
func applySyncCommitReceiptsMigration(ctx context.Context, tx *sql.Tx, _ bool) error {
_, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS sync_commit_receipts (
stage_id TEXT PRIMARY KEY,
checksum TEXT NOT NULL,
repo_id TEXT NOT NULL,
collection TEXT NOT NULL,
committed_at TEXT NOT NULL,
FOREIGN KEY(repo_id) REFERENCES repos(repo_id) ON DELETE CASCADE
)`)
return err
}
func applyMaintenanceLastSuccessMigration(ctx context.Context, tx *sql.Tx, _ bool) error {
columns, err := tableColumns(ctx, tx, "maintenance_frontiers")
if err != nil {
return err
}
if !columns["last_success_at"] {
if _, err := tx.ExecContext(ctx, `ALTER TABLE maintenance_frontiers ADD COLUMN last_success_at TEXT NOT NULL DEFAULT ''`); err != nil {
return err
}
}
_, err = tx.ExecContext(ctx, `UPDATE maintenance_frontiers
SET last_success_at = updated_at
WHERE last_success_at = '' AND last_error_class = '' AND lower(trim(status)) <> 'degraded'`)
return err
}
func tableColumns(ctx context.Context, tx *sql.Tx, table string) (map[string]bool, error) {
rows, err := tx.QueryContext(ctx, `PRAGMA table_info(`+table+`)`)
if err != nil {
return nil, err
}
defer rows.Close()
columns := map[string]bool{}
for rows.Next() {
var cid int
var name, columnType string
var notNull int
var defaultValue any
var pk int
if err := rows.Scan(&cid, &name, &columnType, ¬Null, &defaultValue, &pk); err != nil {
return nil, err
}
columns[name] = true
}
return columns, rows.Err()
}
func detectFTS5(ctx context.Context, db *sql.DB) bool {
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return false
}
defer tx.Rollback()
if _, err := tx.ExecContext(ctx, `CREATE VIRTUAL TABLE temp.fts5_probe USING fts5(value)`); err != nil {
return false
}
return true
}