fn_registry/functions/infra/vault_aggregate_index.go

package infra

import (
	"fmt"
	"os"
	"path/filepath"
	"time"
)

// AggregateReport summarises the result of a VaultAggregateIndex run.
type AggregateReport struct {
	VaultsProcessed int
	VaultsSkipped   int      // vaults without a vault_index.db
	TotalFiles      int
	Errors          []string // non-fatal per-vault errors
}

// VaultAggregateIndex reads all vault manifests from repoRoot, opens each
// vault_index.db and copies all file records into the central registry.db
// vault_files table. The table is created if it does not exist (idempotent).
//
// For each vault the previous rows are deleted and replaced atomically, so
// re-running always produces a clean, non-duplicated state.
//
// Returns an AggregateReport with counts. Per-vault errors are non-fatal
// (logged in report.Errors); only fatal errors (e.g. registry.db
// unreachable) are returned as the error value.
func VaultAggregateIndex(repoRoot string) (AggregateReport, error) {
	var report AggregateReport

	// 1. Open registry.db
	registryDB, err := SQLiteOpen(filepath.Join(repoRoot, "registry.db"), "")
	if err != nil {
		return report, fmt.Errorf("vault_aggregate_index: open registry.db: %w", err)
	}
	defer registryDB.Close()

	// 2. Idempotent schema migration
	for _, stmt := range []string{
		`CREATE TABLE IF NOT EXISTS vault_files (
    vault_id    TEXT NOT NULL,
    vault_name  TEXT NOT NULL,
    rel_path    TEXT NOT NULL,
    size        INTEGER NOT NULL,
    mtime       INTEGER NOT NULL,
    sha256      TEXT NOT NULL,
    mime        TEXT NOT NULL DEFAULT '',
    ext         TEXT NOT NULL DEFAULT '',
    bucket      TEXT NOT NULL DEFAULT '',
    sub_bucket  TEXT NOT NULL DEFAULT '',
    indexed_at  INTEGER NOT NULL,
    PRIMARY KEY (vault_id, rel_path)
);`,
		`CREATE INDEX IF NOT EXISTS idx_vault_files_sha256 ON vault_files(sha256);`,
		`CREATE INDEX IF NOT EXISTS idx_vault_files_vault ON vault_files(vault_id);`,
	} {
		if _, err := registryDB.Exec(stmt); err != nil {
			if !isIdempotentMigrationError(err) {
				return report, fmt.Errorf("vault_aggregate_index: schema: %w", err)
			}
		}
	}

	// 3. Read manifest
	entries, err := VaultManifestRead(repoRoot)
	if err != nil {
		return report, fmt.Errorf("vault_aggregate_index: manifest: %w", err)
	}

	now := time.Now().UTC().Unix()

	for _, entry := range entries {
		vaultID := vaultIDFromEntry(entry)
		vaultName := entry.Name
		vaultPath := entry.Path

		indexPath := filepath.Join(vaultPath, "vault_index.db")
		if _, statErr := os.Stat(indexPath); statErr != nil {
			report.VaultsSkipped++
			continue
		}

		vaultDB, openErr := VaultIndexOpen(vaultPath)
		if openErr != nil {
			report.Errors = append(report.Errors, fmt.Sprintf("%s: open index: %v", vaultName, openErr))
			continue
		}

		rows, queryErr := vaultDB.Query(
			`SELECT rel_path, size, mtime, sha256, mime, ext, bucket, sub_bucket FROM files`,
		)
		if queryErr != nil {
			vaultDB.Close()
			report.Errors = append(report.Errors, fmt.Sprintf("%s: query files: %v", vaultName, queryErr))
			continue
		}

		type fileRow struct {
			RelPath   string
			Size      int64
			Mtime     int64
			Sha256    string
			Mime      string
			Ext       string
			Bucket    string
			SubBucket string
		}
		var fileRows []fileRow
		for rows.Next() {
			var r fileRow
			if scanErr := rows.Scan(&r.RelPath, &r.Size, &r.Mtime, &r.Sha256, &r.Mime, &r.Ext, &r.Bucket, &r.SubBucket); scanErr != nil {
				continue
			}
			fileRows = append(fileRows, r)
		}
		rows.Close()
		vaultDB.Close()

		// Atomic replace in registry.db
		tx, txErr := registryDB.Begin()
		if txErr != nil {
			report.Errors = append(report.Errors, fmt.Sprintf("%s: begin tx: %v", vaultName, txErr))
			continue
		}

		if _, delErr := tx.Exec(`DELETE FROM vault_files WHERE vault_id = ?`, vaultID); delErr != nil {
			tx.Rollback()
			report.Errors = append(report.Errors, fmt.Sprintf("%s: delete: %v", vaultName, delErr))
			continue
		}

		stmt, prepErr := tx.Prepare(`
INSERT INTO vault_files
    (vault_id, vault_name, rel_path, size, mtime, sha256, mime, ext, bucket, sub_bucket, indexed_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
		if prepErr != nil {
			tx.Rollback()
			report.Errors = append(report.Errors, fmt.Sprintf("%s: prepare: %v", vaultName, prepErr))
			continue
		}

		for _, r := range fileRows {
			if _, insErr := stmt.Exec(vaultID, vaultName, r.RelPath, r.Size, r.Mtime, r.Sha256, r.Mime, r.Ext, r.Bucket, r.SubBucket, now); insErr != nil {
				stmt.Close()
				tx.Rollback()
				report.Errors = append(report.Errors, fmt.Sprintf("%s: insert %s: %v", vaultName, r.RelPath, insErr))
				continue
			}
		}
		stmt.Close()

		if commitErr := tx.Commit(); commitErr != nil {
			report.Errors = append(report.Errors, fmt.Sprintf("%s: commit: %v", vaultName, commitErr))
			continue
		}

		report.VaultsProcessed++
		report.TotalFiles += len(fileRows)
	}

	return report, nil
}

// vaultIDFromEntry constructs the canonical vault ID used in registry.db.
// Pattern: "<vault_name>_<project_id>" — consistent with the vaults table.
func vaultIDFromEntry(e VaultManifestEntry) string {
	if e.ProjectID == "" {
		return e.Name
	}
	return e.Name + "_" + e.ProjectID
}