Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
7.1 kB · 223 lines
Go
at sl/gitmirror
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224package db
import ( "context" "database/sql" "fmt" "net/url" "os" "strings" "testing" "time")
func commitActivityTestDB(tb testing.TB) *sql.DB { tb.Helper() databaseURL := os.Getenv("KNOTMIRROR_TEST_DATABASE_URL") if databaseURL == "" { tb.Skip("KNOTMIRROR_TEST_DATABASE_URL is not set") } ctx := context.Background() admin, err := sql.Open("pgx", databaseURL) if err != nil { tb.Fatal(err) } schema := fmt.Sprintf("commit_activity_test_%d", time.Now().UnixNano()) if _, err := admin.ExecContext(ctx, "create schema "+schema); err != nil { admin.Close() tb.Fatal(err) } parsed, err := url.Parse(databaseURL) if err != nil { tb.Fatal(err) } query := parsed.Query() query.Set("search_path", schema) query.Set("timezone", "Australia/Darwin") parsed.RawQuery = query.Encode() database, err := sql.Open("pgx", parsed.String()) if err != nil { tb.Fatal(err) } database.SetMaxOpenConns(4) if err := database.PingContext(ctx); err != nil { tb.Fatal(err) } tb.Cleanup(func() { database.Close() if _, err := admin.ExecContext(context.Background(), "drop schema "+schema+" cascade"); err != nil { tb.Errorf("dropping test schema: %v", err) } admin.Close() }) if _, err := database.ExecContext(ctx, ` create table repos ( repo_did text primary key, state text not null )`); err != nil { tb.Fatal(err) } return database}
func applyCommitActivityMigration(tb testing.TB, database *sql.DB) { tb.Helper() tx, err := database.BeginTx(context.Background(), nil) if err != nil { tb.Fatal(err) } defer tx.Rollback() if err := gitCommitActivity(context.Background(), tx); err != nil { tb.Fatal(err) } if err := tx.Commit(); err != nil { tb.Fatal(err) }}
func TestCommitActivityPostgresLifecycle(t *testing.T) { ctx := context.Background() database := commitActivityTestDB(t) const repoDID = "did:plc:repo" if _, err := database.ExecContext(ctx, `insert into repos (repo_did, state) values ($1, 'active')`, repoDID); err != nil { t.Fatal(err) } applyCommitActivityMigration(t, database)
job, err := ClaimCommitActivityJob(ctx, database) if err != nil || job == nil || job.RepoDid != repoDID { t.Fatalf("initial migration job = %#v, err = %v", job, err) } if err := EnqueueCommitActivity(ctx, database, repoDID); err != nil { t.Fatal(err) }
current := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) commits := []CommitActivity{ {OID: strings.Repeat("a", 40), Committer: "did:plc:actor", Month: current.Format(time.DateOnly)}, {OID: strings.Repeat("b", 40), Committer: "person@example.com", Month: current.Format(time.DateOnly)}, {OID: strings.Repeat("c", 40), Committer: "person@example.com", Month: current.Format(time.DateOnly)}, } inserted, err := SaveCommitActivityBatch(ctx, database, commits) if err != nil || inserted != 3 { t.Fatalf("first batch inserted = %d, err = %v", inserted, err) } inserted, err = SaveCommitActivityBatch(ctx, database, commits) if err != nil || inserted != 0 { t.Fatalf("replayed batch inserted = %d, err = %v", inserted, err) } counts, err := ListCommitCounts( ctx, database, []string{"did:plc:actor", "person@example.com"}, current, current.AddDate(0, 1, 0), 24, ) if err != nil || len(counts) != 1 || counts[0].Commits != 3 { t.Fatalf("aggregated counts = %#v, err = %v", counts, err) }
refs := []CommitRef{{Name: "refs/heads/main", OID: strings.Repeat("d", 40)}} if err := CompleteCommitActivityJob(ctx, database, *job, refs); err != nil { t.Fatal(err) } job, err = ClaimCommitActivityJob(ctx, database) if err != nil || job == nil { t.Fatalf("dirty replay job = %#v, err = %v", job, err) } storedRefs, err := GetCommitActivityRefs(ctx, database, repoDID) if err != nil || len(storedRefs) != 1 || storedRefs[0] != refs[0] { t.Fatalf("stored refs = %#v, err = %v", storedRefs, err) } if err := CompleteCommitActivityJob(ctx, database, *job, refs); err != nil { t.Fatal(err) } job, err = ClaimCommitActivityJob(ctx, database) if err != nil || job != nil { t.Fatalf("finished job = %#v, err = %v", job, err) }}
func TestCommitActivityPostgresPruningAndBounds(t *testing.T) { ctx := context.Background() database := commitActivityTestDB(t) applyCommitActivityMigration(t, database) current := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) old := current.AddDate(0, -24, 0) if _, err := SaveCommitActivityBatch(ctx, database, []CommitActivity{ {OID: strings.Repeat("e", 40), Committer: "did:plc:actor", Month: old.Format(time.DateOnly)}, {OID: strings.Repeat("f", 40), Committer: "did:plc:actor", Month: current.Format(time.DateOnly)}, }); err != nil { t.Fatal(err) } if err := PruneCommitActivity(ctx, database, current.AddDate(0, -23, 0)); err != nil { t.Fatal(err) } var seen, counts int if err := database.QueryRowContext(ctx, `select count(*) from git_seen_commits`).Scan(&seen); err != nil { t.Fatal(err) } if err := database.QueryRowContext(ctx, `select count(*) from git_commit_counts`).Scan(&counts); err != nil { t.Fatal(err) } if seen != 1 || counts != 1 { t.Fatalf("after prune: seen = %d, counts = %d", seen, counts) } if _, err := ListCommitCounts(ctx, database, nil, old, current.AddDate(0, 1, 0), 24); err == nil { t.Fatal("empty committer list was accepted") } if _, err := ListCommitCounts(ctx, database, []string{"did:plc:actor"}, old, current.AddDate(0, 1, 0), 25); err == nil { t.Fatal("limit above API maximum was accepted") }}
func BenchmarkListCommitCounts250kIdentities(b *testing.B) { ctx := context.Background() database := commitActivityTestDB(b) applyCommitActivityMigration(b, database) current := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) if _, err := database.ExecContext(ctx, ` insert into git_commit_counts (committer, month, commits) select 'did:plc:actor' || actor, $1::date - ((actor % 24) || ' months')::interval, 1 from generate_series(1, 250000) actor`, current); err != nil { b.Fatal(err) } committers := make([]string, 0, 251) committers = append(committers, "did:plc:target") for index := range 250 { committers = append(committers, fmt.Sprintf("person%03d@example.com", index)) } if _, err := database.ExecContext(ctx, ` insert into git_commit_counts (committer, month, commits) select committer, $1::date - (month || ' months')::interval, 1 from unnest($2::text[]) committer cross join generate_series(0, 23) month`, current, committers); err != nil { b.Fatal(err) } var relationBytes int64 if err := database.QueryRowContext(ctx, `select pg_total_relation_size('git_commit_counts')`).Scan(&relationBytes); err != nil { b.Fatal(err) } b.ReportMetric(250000+251*24, "dataset_rows") b.ReportMetric(float64(relationBytes), "dataset_bytes") b.ReportAllocs() b.ResetTimer() for range b.N { counts, err := ListCommitCounts( ctx, database, committers, current.AddDate(0, -23, 0), current.AddDate(0, 1, 0), 24, ) if err != nil || len(counts) != 24 { b.Fatalf("counts = %d months, err = %v", len(counts), err) } }}