diff --git a/go.mod b/go.mod index d300f10..0f9d76d 100644 --- a/go.mod +++ b/go.mod @@ -14,6 +14,7 @@ require ( github.com/mattn/go-sqlite3 v1.14.32 github.com/opencontainers/go-digest v1.0.0 github.com/spf13/cobra v1.8.0 + go.yaml.in/yaml/v4 v4.0.0-rc.2 golang.org/x/crypto v0.39.0 ) diff --git a/go.sum b/go.sum index 3170802..cc63040 100644 --- a/go.sum +++ b/go.sum @@ -282,6 +282,8 @@ go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= go.uber.org/zap v1.26.0 h1:sI7k6L95XOKS281NhVKOFCUNIvv9e0w4BF8N3u+tCRo= go.uber.org/zap v1.26.0/go.mod h1:dtElttAiwGvoJ/vj4IwHBS/gXsEu/pZ50mUIRWuG0so= +go.yaml.in/yaml/v4 v4.0.0-rc.2 h1:/FrI8D64VSr4HtGIlUtlFMGsm7H7pWTbj6vOLVZcA6s= +go.yaml.in/yaml/v4 v4.0.0-rc.2/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0= golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.39.0 h1:SHs+kF4LP+f+p14esP5jAoDpHU8Gu/v9lFRK6IT5imM= diff --git a/pkg/appview/db/migrations/0001_remove_star_count_from_repository_stats.yaml b/pkg/appview/db/migrations/0001_remove_star_count_from_repository_stats.yaml new file mode 100644 index 0000000..e929b40 --- /dev/null +++ b/pkg/appview/db/migrations/0001_remove_star_count_from_repository_stats.yaml @@ -0,0 +1,8 @@ +version: 1 +name: remove_star_count_from_repository_stats +up: | + -- Drop star_count column if it exists (SQLite 3.35.0+) + ALTER TABLE repository_stats DROP COLUMN IF EXISTS star_count; + + -- Drop the old star_count index if it exists + DROP INDEX IF EXISTS idx_repository_stats_star_count; diff --git a/pkg/appview/db/migrations/README.md b/pkg/appview/db/migrations/README.md new file mode 100644 index 0000000..5c23bfb --- /dev/null +++ b/pkg/appview/db/migrations/README.md @@ -0,0 +1,77 @@ +# Database Migrations + +This directory contains database migrations for the ATCR AppView database. + +## Migration Format + +Each migration is a YAML file with the following structure: + +```yaml +version: 1 +name: descriptive_migration_name +up: | + SQL commands to apply the migration +``` + +## Naming Convention + +Migration files should be named: `{version:04d}_{name}.yaml` + +Examples: +- `0001_remove_star_count_from_repository_stats.yaml` +- `0002_add_repository_labels.yaml` +- `0003_create_webhooks_table.yaml` + +## Creating a New Migration + +1. **Choose the next version number** - Look at existing migrations and increment by 1 +2. **Create a new YAML file** with the naming convention above +3. **Write your SQL** - Use the `|` block scalar for clean multi-line SQL +4. **Use `IF EXISTS` / `IF NOT EXISTS`** where possible for idempotency + +## Examples + +### Simple single-statement migration: + +```yaml +version: 2 +name: add_repository_description_index +up: | + CREATE INDEX IF NOT EXISTS idx_manifests_description ON manifests(description); +``` + +### Complex multi-statement migration: + +```yaml +version: 3 +name: create_webhooks_table +up: | + -- Create webhooks table + CREATE TABLE IF NOT EXISTS webhooks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + url TEXT NOT NULL, + events TEXT NOT NULL, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + + -- Create index on URL for faster lookups + CREATE INDEX IF NOT EXISTS idx_webhooks_url ON webhooks(url); + + -- Create index on events for filtering + CREATE INDEX IF NOT EXISTS idx_webhooks_events ON webhooks(events); +``` + +## How Migrations Run + +1. Migrations are loaded from this directory on startup +2. Sorted by version number (ascending) +3. Each migration is checked against the `schema_migrations` table +4. Only unapplied migrations are executed +5. After successful execution, the version is recorded in `schema_migrations` + +## Important Notes + +- **Never modify existing migrations** - Once applied, they're immutable +- **Test migrations** before committing - Ensure they work on existing databases +- **Version numbers must be unique** - The migration system will fail if duplicates exist +- **Migrations are run automatically** on `InitDB()` - No manual intervention needed diff --git a/pkg/appview/db/models.go b/pkg/appview/db/models.go index d8d4c21..148d305 100644 --- a/pkg/appview/db/models.go +++ b/pkg/appview/db/models.go @@ -81,7 +81,7 @@ type Repository struct { type RepositoryStats struct { DID string `json:"did"` Repository string `json:"repository"` - StarCount int `json:"star_count"` + StarCount int `json:"star_count"` // Calculated from stars table, not stored PullCount int `json:"pull_count"` LastPull *time.Time `json:"last_pull,omitempty"` PushCount int `json:"push_count"` diff --git a/pkg/appview/db/queries.go b/pkg/appview/db/queries.go index 3fb5a32..a98b497 100644 --- a/pkg/appview/db/queries.go +++ b/pkg/appview/db/queries.go @@ -767,11 +767,19 @@ func GetRepositoryStats(db *sql.DB, did, repository string) (*RepositoryStats, e var stats RepositoryStats var lastPullStr, lastPushStr sql.NullString + // Get pull/push stats from repository_stats, and star count from stars table err := db.QueryRow(` - SELECT did, repository, star_count, pull_count, last_pull, push_count, last_push - FROM repository_stats - WHERE did = ? AND repository = ? - `, did, repository).Scan(&stats.DID, &stats.Repository, &stats.StarCount, &stats.PullCount, &lastPullStr, &stats.PushCount, &lastPushStr) + SELECT + COALESCE(rs.did, ?) as did, + COALESCE(rs.repository, ?) as repository, + (SELECT COUNT(*) FROM stars WHERE owner_did = ? AND repository = ?) as star_count, + COALESCE(rs.pull_count, 0) as pull_count, + rs.last_pull, + COALESCE(rs.push_count, 0) as push_count, + rs.last_push + FROM (SELECT ? as did, ? as repository) AS placeholder + LEFT JOIN repository_stats rs ON rs.did = ? AND rs.repository = ? + `, did, repository, did, repository, did, repository, did, repository).Scan(&stats.DID, &stats.Repository, &stats.StarCount, &stats.PullCount, &lastPullStr, &stats.PushCount, &lastPushStr) if err == sql.ErrNoRows { // Return zero stats if no record exists yet @@ -840,6 +848,103 @@ func DecrementStarCount(db *sql.DB, did, repository string) error { return err } +// UpsertStar inserts or updates a star record (idempotent) +func UpsertStar(db *sql.DB, starrerDID, ownerDID, repository string, createdAt time.Time) error { + _, err := db.Exec(` + INSERT INTO stars (starrer_did, owner_did, repository, created_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(starrer_did, owner_did, repository) DO UPDATE SET + created_at = excluded.created_at + `, starrerDID, ownerDID, repository, createdAt) + return err +} + +// DeleteStar deletes a star record +func DeleteStar(db *sql.DB, starrerDID, ownerDID, repository string) error { + _, err := db.Exec(` + DELETE FROM stars + WHERE starrer_did = ? AND owner_did = ? AND repository = ? + `, starrerDID, ownerDID, repository) + return err +} + +// RebuildStarCount rebuilds the star count for a specific repository from the stars table +func RebuildStarCount(db *sql.DB, ownerDID, repository string) error { + _, err := db.Exec(` + INSERT INTO repository_stats (did, repository, star_count) + VALUES (?, ?, ( + SELECT COUNT(*) FROM stars + WHERE owner_did = ? AND repository = ? + )) + ON CONFLICT(did, repository) DO UPDATE SET + star_count = ( + SELECT COUNT(*) FROM stars + WHERE owner_did = ? AND repository = ? + ) + `, ownerDID, repository, ownerDID, repository, ownerDID, repository) + return err +} + +// GetStarsForDID returns all stars created by a specific DID (for backfill reconciliation) +// Returns a map of (ownerDID, repository) -> createdAt +func GetStarsForDID(db *sql.DB, starrerDID string) (map[string]time.Time, error) { + rows, err := db.Query(` + SELECT owner_did, repository, created_at + FROM stars + WHERE starrer_did = ? + `, starrerDID) + if err != nil { + return nil, err + } + defer rows.Close() + + stars := make(map[string]time.Time) + for rows.Next() { + var ownerDID, repository string + var createdAt time.Time + if err := rows.Scan(&ownerDID, &repository, &createdAt); err != nil { + return nil, err + } + key := fmt.Sprintf("%s/%s", ownerDID, repository) + stars[key] = createdAt + } + + return stars, rows.Err() +} + +// DeleteStarsNotInList deletes stars from the database that are not in the provided list +// This is used during backfill reconciliation to remove stars that no longer exist on PDS +func DeleteStarsNotInList(db *sql.DB, starrerDID string, foundStars map[string]time.Time) error { + // Get current stars in DB + currentStars, err := GetStarsForDID(db, starrerDID) + if err != nil { + return fmt.Errorf("failed to get current stars: %w", err) + } + + // Find stars to delete (in DB but not on PDS) + var toDelete []struct{ ownerDID, repository string } + for key := range currentStars { + if _, exists := foundStars[key]; !exists { + parts := strings.SplitN(key, "/", 2) + if len(parts) == 2 { + toDelete = append(toDelete, struct{ ownerDID, repository string }{ + ownerDID: parts[0], + repository: parts[1], + }) + } + } + } + + // Delete orphaned stars + for _, star := range toDelete { + if err := DeleteStar(db, starrerDID, star.ownerDID, star.repository); err != nil { + return fmt.Errorf("failed to delete star: %w", err) + } + } + + return nil +} + // IncrementPullCount increments the pull count for a repository func IncrementPullCount(db *sql.DB, did, repository string) error { _, err := db.Exec(` diff --git a/pkg/appview/db/schema.go b/pkg/appview/db/schema.go index c583e53..f11f84c 100644 --- a/pkg/appview/db/schema.go +++ b/pkg/appview/db/schema.go @@ -2,11 +2,21 @@ package db import ( "database/sql" + "fmt" + "os" + "path/filepath" + "sort" _ "github.com/mattn/go-sqlite3" + "go.yaml.in/yaml/v4" ) const schema = ` +CREATE TABLE IF NOT EXISTS schema_migrations ( + version INTEGER PRIMARY KEY, + applied_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP +); + CREATE TABLE IF NOT EXISTS users ( did TEXT PRIMARY KEY, handle TEXT NOT NULL, @@ -144,7 +154,6 @@ CREATE INDEX IF NOT EXISTS idx_pending_device_auth_expires ON pending_device_aut CREATE TABLE IF NOT EXISTS repository_stats ( did TEXT NOT NULL, repository TEXT NOT NULL, - star_count INTEGER NOT NULL DEFAULT 0, pull_count INTEGER NOT NULL DEFAULT 0, last_pull TIMESTAMP, push_count INTEGER NOT NULL DEFAULT 0, @@ -153,8 +162,19 @@ CREATE TABLE IF NOT EXISTS repository_stats ( FOREIGN KEY(did) REFERENCES users(did) ON DELETE CASCADE ); CREATE INDEX IF NOT EXISTS idx_repository_stats_did ON repository_stats(did); -CREATE INDEX IF NOT EXISTS idx_repository_stats_star_count ON repository_stats(star_count DESC); CREATE INDEX IF NOT EXISTS idx_repository_stats_pull_count ON repository_stats(pull_count DESC); + +CREATE TABLE IF NOT EXISTS stars ( + starrer_did TEXT NOT NULL, + owner_did TEXT NOT NULL, + repository TEXT NOT NULL, + created_at TIMESTAMP NOT NULL, + PRIMARY KEY(starrer_did, owner_did, repository), + FOREIGN KEY(starrer_did) REFERENCES users(did) ON DELETE CASCADE, + FOREIGN KEY(owner_did) REFERENCES users(did) ON DELETE CASCADE +); +CREATE INDEX IF NOT EXISTS idx_stars_owner_repo ON stars(owner_did, repository); +CREATE INDEX IF NOT EXISTS idx_stars_starrer ON stars(starrer_did); ` // InitDB initializes the SQLite database with the schema @@ -174,5 +194,105 @@ func InitDB(path string) (*sql.DB, error) { return nil, err } + // Run migrations + if err := runMigrations(db); err != nil { + return nil, err + } + return db, nil +} + +// Migration represents a database migration +type Migration struct { + Version int `yaml:"version"` + Name string `yaml:"name"` + Up string `yaml:"up"` +} + +// runMigrations applies any pending database migrations +func runMigrations(db *sql.DB) error { + // Load migrations from files + migrations, err := loadMigrations() + if err != nil { + return fmt.Errorf("failed to load migrations: %w", err) + } + + // Sort migrations by version + sort.Slice(migrations, func(i, j int) bool { + return migrations[i].Version < migrations[j].Version + }) + + for _, m := range migrations { + // Check if migration already applied + var count int + err := db.QueryRow("SELECT COUNT(*) FROM schema_migrations WHERE version = ?", m.Version).Scan(&count) + if err != nil { + return fmt.Errorf("failed to check migration status: %w", err) + } + + if count > 0 { + // Migration already applied + continue + } + + // Apply migration + fmt.Printf("Applying migration %d: %s\n", m.Version, m.Name) + if _, err := db.Exec(m.Up); err != nil { + return fmt.Errorf("failed to apply migration %d (%s): %w", m.Version, m.Name, err) + } + + // Record migration + if _, err := db.Exec("INSERT INTO schema_migrations (version) VALUES (?)", m.Version); err != nil { + return fmt.Errorf("failed to record migration %d: %w", m.Version, err) + } + + fmt.Printf("Migration %d applied successfully\n", m.Version) + } + + return nil +} + +// loadMigrations loads all migration files from the migrations directory +func loadMigrations() ([]Migration, error) { + // Get the path to the migrations directory + // Try relative to working directory first, then relative to this file + migrationsDir := "pkg/appview/db/migrations" + if _, err := os.Stat(migrationsDir); os.IsNotExist(err) { + // Try embedded path (when running from different directory) + migrationsDir = filepath.Join(".", "migrations") + } + + // Read all .yaml files in the migrations directory + files, err := filepath.Glob(filepath.Join(migrationsDir, "*.yaml")) + if err != nil { + return nil, fmt.Errorf("failed to list migration files: %w", err) + } + + var migrations []Migration + for _, file := range files { + data, err := os.ReadFile(file) + if err != nil { + return nil, fmt.Errorf("failed to read migration file %s: %w", file, err) + } + + var m Migration + if err := yaml.Unmarshal(data, &m); err != nil { + return nil, fmt.Errorf("failed to parse migration file %s: %w", file, err) + } + + // Validate migration + if m.Version <= 0 { + return nil, fmt.Errorf("invalid migration version in %s: %d", file, m.Version) + } + if m.Name == "" { + return nil, fmt.Errorf("missing migration name in %s", file) + } + if m.Up == "" { + return nil, fmt.Errorf("missing migration 'up' SQL in %s", file) + } + + migrations = append(migrations, m) + } + + return migrations, nil } \ No newline at end of file diff --git a/pkg/appview/jetstream/backfill.go b/pkg/appview/jetstream/backfill.go index e88dc8d..6fdaf5c 100644 --- a/pkg/appview/jetstream/backfill.go +++ b/pkg/appview/jetstream/backfill.go @@ -145,6 +145,7 @@ func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection strin // Track which records exist on the PDS for reconciliation var foundManifestDigests []string var foundTags []struct{ Repository, Tag string } + foundStars := make(map[string]time.Time) // key: "ownerDID/repository", value: createdAt // Paginate through all records for this repo for { @@ -169,6 +170,12 @@ func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection strin Tag: tagRecord.Tag, }) } + } else if collection == atproto.StarCollection { + var starRecord atproto.StarRecord + if err := json.Unmarshal(record.Value, &starRecord); err == nil { + key := fmt.Sprintf("%s/%s", starRecord.Subject.DID, starRecord.Subject.Repository) + foundStars[key] = starRecord.CreatedAt + } } if err := b.processRecord(ctx, did, collection, &record); err != nil { @@ -187,7 +194,7 @@ func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection strin } // Reconcile deletions - remove records from DB that no longer exist on PDS - if err := b.reconcileDeletions(did, collection, foundManifestDigests, foundTags); err != nil { + if err := b.reconcileDeletions(did, collection, foundManifestDigests, foundTags, foundStars); err != nil { fmt.Printf("WARNING: Failed to reconcile deletions for %s: %v\n", did, err) } @@ -195,7 +202,7 @@ func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection strin } // reconcileDeletions removes records from the database that no longer exist on the PDS -func (b *BackfillWorker) reconcileDeletions(did, collection string, foundManifestDigests []string, foundTags []struct{ Repository, Tag string }) error { +func (b *BackfillWorker) reconcileDeletions(did, collection string, foundManifestDigests []string, foundTags []struct{ Repository, Tag string }, foundStars map[string]time.Time) error { switch collection { case atproto.ManifestCollection: // Get current manifests in DB @@ -232,6 +239,13 @@ func (b *BackfillWorker) reconcileDeletions(did, collection string, foundManifes if deleted > 0 { fmt.Printf("Backfill: Deleted %d orphaned tags for %s\n", deleted, did) } + + case atproto.StarCollection: + // Reconcile stars - delete stars that no longer exist on PDS + // Star counts will be calculated on demand from the stars table + if err := db.DeleteStarsNotInList(b.db, did, foundStars); err != nil { + return fmt.Errorf("failed to delete orphaned stars: %w", err) + } } return nil @@ -336,10 +350,11 @@ func (b *BackfillWorker) processStarRecord(did string, record *atproto.Record) e return fmt.Errorf("failed to unmarshal star: %w", err) } - // Increment star count for the repository being starred + // Upsert the star record (idempotent - won't duplicate) // The DID here is the starrer (user who starred) // The subject contains the owner DID and repository - return db.IncrementStarCount(b.db, starRecord.Subject.DID, starRecord.Subject.Repository) + // Star count will be calculated on demand from the stars table + return db.UpsertStar(b.db, did, starRecord.Subject.DID, starRecord.Subject.Repository, starRecord.CreatedAt) } // ensureUser resolves and upserts a user by DID diff --git a/pkg/appview/jetstream/worker.go b/pkg/appview/jetstream/worker.go index f613c2b..65148df 100644 --- a/pkg/appview/jetstream/worker.go +++ b/pkg/appview/jetstream/worker.go @@ -415,22 +415,15 @@ func (w *Worker) processStar(commit *CommitEvent) error { } if commit.Operation == "delete" { - // Unstar - parse the record to get the subject (owner DID and repository) - var starRecord atproto.StarRecord - if commit.Record != nil { - recordBytes, err := json.Marshal(commit.Record) - if err != nil { - return fmt.Errorf("failed to marshal record: %w", err) - } - if err := json.Unmarshal(recordBytes, &starRecord); err != nil { - return fmt.Errorf("failed to unmarshal star: %w", err) - } - - // Decrement star count - return db.DecrementStarCount(w.db, starRecord.Subject.DID, starRecord.Subject.Repository) + // Unstar - parse the rkey to get the subject (owner DID and repository) + // Delete events don't include the full record, but the rkey contains the info we need + ownerDID, repository, err := atproto.ParseStarRecordKey(commit.RKey) + if err != nil { + return fmt.Errorf("failed to parse star rkey: %w", err) } - // If no record data, we can't determine what was unstarred - return nil + + // Delete the star record + return db.DeleteStar(w.db, commit.DID, ownerDID, repository) } // Parse star record @@ -447,8 +440,8 @@ func (w *Worker) processStar(commit *CommitEvent) error { return nil } - // Increment star count for the repository being starred - return db.IncrementStarCount(w.db, starRecord.Subject.DID, starRecord.Subject.Repository) + // Upsert the star record (idempotent - star count will be calculated on demand) + return db.UpsertStar(w.db, commit.DID, starRecord.Subject.DID, starRecord.Subject.Repository, starRecord.CreatedAt) } // JetstreamEvent represents a Jetstream event diff --git a/pkg/appview/static/js/app.js b/pkg/appview/static/js/app.js index 9d13d82..50eb0c2 100644 --- a/pkg/appview/static/js/app.js +++ b/pkg/appview/static/js/app.js @@ -172,8 +172,8 @@ async function toggleStar(handle, repository) { starCountEl.textContent = Math.max(0, currentCount - 1); } - // Refresh actual count from server (will correct if optimistic update was wrong) - await loadStarCount(handle, repository); + // Don't fetch count immediately - trust the optimistic update + // The actual count will be correct on next page load } catch (err) { console.error('Error toggling star:', err); diff --git a/pkg/atproto/lexicon.go b/pkg/atproto/lexicon.go index 87ebbc6..607c229 100644 --- a/pkg/atproto/lexicon.go +++ b/pkg/atproto/lexicon.go @@ -3,6 +3,8 @@ package atproto import ( "encoding/base64" "encoding/json" + "fmt" + "strings" "time" ) @@ -302,3 +304,18 @@ func StarRecordKey(ownerDID, repository string) string { combined := ownerDID + "/" + repository return base64.RawURLEncoding.EncodeToString([]byte(combined)) } + +// ParseStarRecordKey decodes a star record key back to ownerDID and repository +func ParseStarRecordKey(rkey string) (ownerDID, repository string, err error) { + decoded, err := base64.RawURLEncoding.DecodeString(rkey) + if err != nil { + return "", "", fmt.Errorf("failed to decode star rkey: %w", err) + } + + parts := strings.SplitN(string(decoded), "/", 2) + if len(parts) != 2 { + return "", "", fmt.Errorf("invalid star rkey format: %s", string(decoded)) + } + + return parts[0], parts[1], nil +}