diff --git a/.gitignore b/.gitignore index 595dfa4..4e1454c 100644 --- a/.gitignore +++ b/.gitignore @@ -12,3 +12,6 @@ static/*.db-wal .secrets .env + +internal-docs/ +internal/devseed/fixtures/ diff --git a/README.md b/README.md index 1926b7f..bb013e9 100644 --- a/README.md +++ b/README.md @@ -47,4 +47,23 @@ API keys are scoped to an entity and carry a permission bitfield. Keys are prefi ## Rate Limiting -Rate limits are enforced in-memory per process. Throttled responses return `429 Too Many Requests` with `X-RateLimit-*` and `Retry-After` headers. \ No newline at end of file +Rate limits are enforced in-memory per process. Throttled responses return `429 Too Many Requests` with `X-RateLimit-*` and `Retry-After` headers. + +## Seeding local data + +Production ingests live data from contributing marketplaces. For local development, a synthetic dataset (~6 months, ~2k rows with realistic daily variance) can be loaded with: + +```bash +go run ./cmd/seed +``` + +The seed: + +- Must be run in a development environment against a local database. +- Generates a 6-month dataset ending today, so the dashboard at `/` has enough data to exercise every period (7d / 30d / 3m / 6m / 1y). +- Uses Steam IDs from the unallocated top of the 32-bit account space (`76561201986797568`–`76561202255233023`) and `dev-`-prefixed marketplace slugs, so synthetic rows never collide with real accounts or real ingestors. +- Skips rows that already exist, so a plain rerun inserts nothing. To refresh the data so it ends at today, delete the existing synthetic rows first: + +```bash +go run ./cmd/seed -reset +``` \ No newline at end of file diff --git a/cmd/seed/main.go b/cmd/seed/main.go new file mode 100644 index 0000000..b9adf66 --- /dev/null +++ b/cmd/seed/main.go @@ -0,0 +1,107 @@ +// Command seed loads a synthetic dataset into the local Reverse Watch +// Postgres databases for local dashboard development. It is NEVER intended +// to run in production: it refuses to run unless Config.Environment is +// "development" and Database.Host points at a local database. +// +// go run ./cmd/seed # insert; rows that already exist are skipped +// go run ./cmd/seed -reset # delete existing synthetic rows first, then insert +package main + +import ( + "flag" + "fmt" + "net" + "os" + "strings" + "time" + + "reverse-watch/config" + "reverse-watch/domain/models" + "reverse-watch/domain/models/constants" + "reverse-watch/internal/devseed" + "reverse-watch/logging" + "reverse-watch/repository/factory" + "reverse-watch/secret" + + "gorm.io/gorm" + "gorm.io/gorm/logger" +) + +func main() { + reset := flag.Bool("reset", false, "delete existing synthetic reversals before seeding, so a rerun refreshes the data to end at today") + flag.Parse() + + if err := run(*reset); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func run(reset bool) error { + logging.Initialize() + cfg := config.Load() + + if cfg.Environment != constants.EnvironmentDevelopment { + return fmt.Errorf("refusing to seed: environment is %q (only %q is allowed)", cfg.Environment, constants.EnvironmentDevelopment) + } + if !isLocalHost(cfg.Database.Host) { + return fmt.Errorf("refusing to seed: database host %q is not local (expected localhost, a loopback address, a unix socket, or a docker service name)", cfg.Database.Host) + } + + // Required by factory bootstrap (e.g. admin API key seeding). The + // synthetic generator pre-populates its own IDs, so the snowflake + // generator does not actually run for them. + models.InitSnowflakeGenerator(0, 0) + + keygen := secret.NewKeyGenerator(cfg.Environment) + f, err := factory.NewFactory(cfg, keygen) + if err != nil { + return fmt.Errorf("failed to create factory: %w", err) + } + defer func() { + if err := f.Close(); err != nil { + fmt.Fprintf(os.Stderr, "failed to close factory: %v\n", err) + } + }() + + // The factory logs every statement at Info level, which would dump the + // full SQL of each 1,000-row batch. + db := f.PublicDB().Session(&gorm.Session{Logger: logger.Default.LogMode(logger.Warn)}) + + if reset { + deleted, err := devseed.ResetSynthetic(db) + if err != nil { + return fmt.Errorf("failed to reset synthetic reversals: %w", err) + } + fmt.Printf("reset: deleted %d existing synthetic reversals\n", deleted) + } + + reversals := devseed.GenerateSynthetic(time.Now().UTC()) + fmt.Printf("generated %d synthetic reversals\n", len(reversals)) + + inserted, err := devseed.InsertReversals(db, reversals) + if err != nil { + return fmt.Errorf("failed to insert reversals: %w", err) + } + skipped := int64(len(reversals)) - inserted + fmt.Printf("seed complete: %d inserted, %d already present (skipped)\n", inserted, skipped) + if skipped > 0 && !reset { + fmt.Println("hint: run with -reset to replace existing synthetic data with a fresh dataset ending today") + } + return nil +} + +// isLocalHost accepts loopback hosts, unix socket directories, and +// single-label hostnames (docker compose service names such as "postgres"). +// Anything else with a dot or colon (FQDNs, non-loopback IPs, Cloud SQL +// connection names like "project:region:instance") is treated as remote. +func isLocalHost(host string) bool { + host = strings.TrimSpace(host) + if host == "" || host == "localhost" || strings.HasPrefix(host, "/") { + return true + } + if ip := net.ParseIP(host); ip != nil { + return ip.IsLoopback() + } + return !strings.ContainsAny(host, ".:") +} diff --git a/cmd/seed/main_test.go b/cmd/seed/main_test.go new file mode 100644 index 0000000..c7cb595 --- /dev/null +++ b/cmd/seed/main_test.go @@ -0,0 +1,28 @@ +package main + +import "testing" + +func TestIsLocalHost(t *testing.T) { + testCases := []struct { + host string + want bool + }{ + {host: "", want: true}, + {host: "localhost", want: true}, + {host: "127.0.0.1", want: true}, + {host: "::1", want: true}, + {host: "/var/run/postgresql", want: true}, + {host: "postgres", want: true}, + {host: "db", want: true}, + {host: "10.0.0.5", want: false}, + {host: "db.example.com", want: false}, + {host: "my-project:us-central1:reverse-watch", want: false}, + } + for _, tc := range testCases { + t.Run(tc.host, func(t *testing.T) { + if got := isLocalHost(tc.host); got != tc.want { + t.Errorf("isLocalHost(%q) = %v, want %v", tc.host, got, tc.want) + } + }) + } +} diff --git a/internal/devseed/synthetic.go b/internal/devseed/synthetic.go new file mode 100644 index 0000000..16dc8cf --- /dev/null +++ b/internal/devseed/synthetic.go @@ -0,0 +1,212 @@ +// Package devseed loads dev-only fixture data into the local Postgres +// instance. It is intentionally not wired into the main binary — call it +// from cmd/seed (or a test) when you need realistic data locally. +package devseed + +import ( + "math" + "math/rand" + "time" + + "reverse-watch/domain/models" + + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +// Synthetic Steam IDs live at the top of the 32-bit account space +// (account IDs >= 0xF0000000), which Valve has not allocated. They still +// pass SteamID.IsValid, and the dedicated range lets ResetSynthetic delete +// exactly the rows this package generated. +const ( + steamID64Base uint64 = 76561197960265728 + SyntheticSteamIDMin uint64 = steamID64Base + 0xF0000000 + SyntheticSteamIDMax uint64 = steamID64Base + 0xFFFFFFFF + syntheticRNGSeed int64 = 42 + syntheticDays int = 180 + syntheticTargetTotal int = 2000 +) + +// Marketplace slugs are prefixed with "dev-" so synthetic rows never match a +// real ingestor's slug. The CSFloat ingestor resumes from the highest +// reporter_internal_id on "csfloat" rows, so sharing that slug would corrupt +// its cursor. +var syntheticMarketplaces = []struct { + slug string + weight float64 +}{ + {"dev-csfloat", 0.80}, + {"dev-tradeit", 0.10}, + {"dev-skinport", 0.05}, + {"dev-swap.gg", 0.05}, +} + +// GenerateSynthetic returns a ~6-month dataset (~2,000 rows, at least one +// per day, gentle sinusoid with occasional spikes / quiet days) ending at +// now. Snowflake IDs are unique within the slice, so callers can pipe the +// result straight into InsertReversals. +func GenerateSynthetic(now time.Time) []*models.Reversal { + now = now.UTC() + rng := rand.New(rand.NewSource(syntheticRNGSeed)) + today := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC) + + counts := make([]int, syntheticDays) + for d := 0; d < syntheticDays; d++ { + base := 40.0 + 10.0*math.Sin(float64(d)/30.0) + var mult float64 + switch r := rng.Float64(); { + case r < 0.05: + mult = 2.5 + rng.Float64()*2.5 + case r < 0.15: + mult = 0.2 + rng.Float64()*0.3 + default: + mult = 0.7 + rng.Float64()*0.6 + } + counts[d] = int(math.Max(1, base*mult+rng.NormFloat64()*5)) + } + + total := 0 + for _, c := range counts { + total += c + } + if total > 0 { + scale := float64(syntheticTargetTotal) / float64(total) + for d := range counts { + counts[d] = int(math.Max(1, math.Round(float64(counts[d])*scale))) + } + } + + rows := make([]*models.Reversal, 0, syntheticTargetTotal+200) + var steamOffset uint64 = 1 + var seq uint16 + + for d := 0; d < syntheticDays; d++ { + dayStart := today.AddDate(0, 0, -(syntheticDays - 1 - d)) + dayLen := 24 * time.Hour + if dayEnd := dayStart.Add(dayLen); dayEnd.After(now) { + // Today is only partially elapsed: spread rows over [dayStart, now) + // so they don't pile up on a single clamped timestamp. + dayLen = now.Sub(dayStart) + } + for i := 0; i < counts[d]; i++ { + reversedAt := dayStart.Add(time.Duration(rng.Float64() * float64(dayLen))) + createdAt := reversedAt.Add(time.Duration(rng.Intn(8*60)+1) * time.Minute) + if createdAt.After(now) { + createdAt = reversedAt.Add(time.Duration(rng.Float64() * float64(now.Sub(reversedAt)))) + } + + srcRoll := rng.Float64() + var src models.Source + var related *models.SteamID + switch { + case srcRoll < 0.90: + src = models.SourceDirect + case srcRoll < 0.95: + src = models.SourceRelatedUser + relID := models.SteamID(SyntheticSteamIDMin + (steamOffset+10_000)*97) + related = &relID + default: + src = models.SourceUserReport + } + + var expunged *uint64 + if rng.Float64() < 0.015 { + eAt := createdAt.Add(time.Duration(rng.Intn(24*60)+1) * time.Minute) + if eAt.Before(now) { + ems := uint64(eAt.UnixMilli()) + expunged = &ems + } + } + + steamID := models.SteamID(SyntheticSteamIDMin + steamOffset*73 + uint64(rng.Intn(50))) + steamOffset++ + + // Snowflake encodes created_at + a 12-bit per-ms sequence; + // mirrors domain/models/snowflake.go so generated IDs sort + // chronologically alongside production rows. + seq = (seq + 1) & 0x0FFF + // Clamp so a createdAt before models.Epoch can't underflow the + // unsigned subtraction (mirrors the guard in genSnowflakeWithParts). + createdAtMs := uint64(createdAt.UnixMilli()) + if createdAtMs < models.Epoch { + createdAtMs = models.Epoch + } + sf := models.Snowflake(((createdAtMs - models.Epoch) << 22) | uint64(seq)) + + rows = append(rows, &models.Reversal{ + Model: models.Model{ + ID: sf, + CreatedAt: uint64(createdAt.UnixMilli()), + UpdatedAt: uint64(createdAt.UnixMilli()), + }, + SteamID: steamID, + MarketplaceSlug: pickMarketplace(rng), + Source: &src, + RelatedSteamID: related, + ReversedAt: uint64(reversedAt.UnixMilli()), + ExpungedAt: expunged, + }) + } + } + return rows +} + +func pickMarketplace(rng *rand.Rand) string { + r := rng.Float64() + cum := 0.0 + for _, mp := range syntheticMarketplaces { + cum += mp.weight + if r < cum { + return mp.slug + } + } + return syntheticMarketplaces[0].slug +} + +// insertChunkSize keeps each bulk insert under Postgres's 65,535 +// parameter-per-statement cap. At ~11 columns per Reversal, 1,000 rows +// uses ~11k parameters. +const insertChunkSize = 1000 + +// InsertReversals bulk-inserts reversals, skipping any row whose +// (steam_id, marketplace_slug) already exists via ON CONFLICT DO NOTHING. +// The conflict target matches the partial unique index created in +// repository/public (idx_reversals_steam_id_marketplace_slug ... WHERE +// deleted_at IS NULL). Those pairs don't depend on the clock, so a rerun +// without ResetSynthetic inserts nothing. +// +// Only that index is covered: a primary-key collision on id still fails the +// whole batch. Snowflakes embed created_at plus a sequence, so this would +// require two seeds generating the same millisecond and sequence. +// +// Returns the number of rows actually inserted. +func InsertReversals(db *gorm.DB, reversals []*models.Reversal) (int64, error) { + if len(reversals) == 0 { + return 0, nil + } + onConflict := clause.OnConflict{ + Columns: []clause.Column{{Name: "steam_id"}, {Name: "marketplace_slug"}}, + TargetWhere: clause.Where{Exprs: []clause.Expression{clause.Expr{SQL: "deleted_at IS NULL"}}}, + DoNothing: true, + } + var inserted int64 + for i := 0; i < len(reversals); i += insertChunkSize { + end := min(i+insertChunkSize, len(reversals)) + res := db.Clauses(onConflict).Create(reversals[i:end]) + if res.Error != nil { + return inserted, res.Error + } + inserted += res.RowsAffected + } + return inserted, nil +} + +// ResetSynthetic hard-deletes every reversal whose steam_id falls in the +// synthetic range, including soft-deleted rows. Returns the number of rows +// removed. +func ResetSynthetic(db *gorm.DB) (int64, error) { + res := db.Unscoped(). + Where("steam_id BETWEEN ? AND ?", SyntheticSteamIDMin, SyntheticSteamIDMax). + Delete(&models.Reversal{}) + return res.RowsAffected, res.Error +} diff --git a/internal/devseed/synthetic_test.go b/internal/devseed/synthetic_test.go new file mode 100644 index 0000000..086d144 --- /dev/null +++ b/internal/devseed/synthetic_test.go @@ -0,0 +1,163 @@ +package devseed + +import ( + "testing" + "time" + + "reverse-watch/domain/models" + "reverse-watch/internal/testutil" + "reverse-watch/repository/public" +) + +// Re-running the seed produces new snowflake IDs but the same +// (steam_id, marketplace_slug) pairs, which collide with the partial unique +// index. The insert must skip those rows via ON CONFLICT DO NOTHING rather +// than raising a unique-constraint error. +func TestInsertReversals_RerunIsSafe(t *testing.T) { + db := testutil.NewTestDB(t) + if err := public.CreateIndexes(db); err != nil { + t.Fatalf("CreateIndexes: %v", err) + } + + // Both runs use past timestamps (BeforeCreate rejects future + // reversed_at), but on different calendar days so snowflake IDs differ. + now := time.Now().UTC().AddDate(0, 0, -10) + if _, err := InsertReversals(db, GenerateSynthetic(now)); err != nil { + t.Fatalf("first insert: %v", err) + } + + n2, err := InsertReversals(db, GenerateSynthetic(now.AddDate(0, 0, 5))) + if err != nil { + t.Fatalf("rerun errored, expected ON CONFLICT DO NOTHING to skip: %v", err) + } + if n2 != 0 { + t.Errorf("rerun inserted %d rows, want 0", n2) + } +} + +// ResetSynthetic must remove every synthetic row, including soft-deleted +// ones, so a reseed can insert a fresh dataset ending at the new date. Rows +// outside the synthetic Steam ID range must be left alone. +func TestResetSynthetic_RefreshesData(t *testing.T) { + db := testutil.NewTestDB(t) + if err := public.CreateIndexes(db); err != nil { + t.Fatalf("CreateIndexes: %v", err) + } + + nonSynthetic := &models.Reversal{ + Model: models.Model{ID: 1}, + SteamID: models.SteamID(76561197960287930), + MarketplaceSlug: "csfloat", + } + testutil.Insert(t, db, nonSynthetic) + + old := time.Now().UTC().AddDate(0, 0, -10) + first := GenerateSynthetic(old) + if _, err := InsertReversals(db, first); err != nil { + t.Fatalf("first insert: %v", err) + } + if err := db.Delete(&models.Reversal{}, first[0].ID).Error; err != nil { + t.Fatalf("soft delete: %v", err) + } + + deleted, err := ResetSynthetic(db) + if err != nil { + t.Fatalf("ResetSynthetic: %v", err) + } + if deleted != int64(len(first)) { + t.Errorf("ResetSynthetic deleted %d rows, want %d", deleted, len(first)) + } + + fresh := GenerateSynthetic(time.Now().UTC()) + inserted, err := InsertReversals(db, fresh) + if err != nil { + t.Fatalf("reseed: %v", err) + } + if inserted != int64(len(fresh)) { + t.Errorf("reseed inserted %d rows, want %d", inserted, len(fresh)) + } + + var realCount int64 + if err := db.Model(&models.Reversal{}).Where("id = ?", nonSynthetic.ID).Count(&realCount).Error; err != nil { + t.Fatalf("count real row: %v", err) + } + if realCount != 1 { + t.Errorf("non-synthetic row was deleted by ResetSynthetic") + } +} + +// Synthetic rows must use unallocated Steam IDs and must never share a slug +// or reporter ID with a real ingestor, otherwise they would corrupt the +// CSFloat ingestor's resume cursor. +func TestGenerateSynthetic_IsolatedFromRealData(t *testing.T) { + now := time.Now().UTC() + rows := GenerateSynthetic(now) + for _, r := range rows { + ids := []models.SteamID{r.SteamID} + if r.RelatedSteamID != nil { + ids = append(ids, *r.RelatedSteamID) + } + for _, id := range ids { + if !id.IsValid() { + t.Fatalf("steam id %d is not valid", id) + } + if uint64(id) < SyntheticSteamIDMin || uint64(id) > SyntheticSteamIDMax { + t.Fatalf("steam id %d is outside the synthetic range", id) + } + } + if r.MarketplaceSlug == "csfloat" { + t.Fatalf("row %d uses the real csfloat slug", r.ID) + } + if r.ReporterInternalID != nil { + t.Fatalf("row %d has a reporter_internal_id", r.ID) + } + if r.ReversedAt > uint64(now.UnixMilli()) || r.CreatedAt > uint64(now.UnixMilli()) { + t.Fatalf("row %d has a timestamp after now", r.ID) + } + } +} + +// Seeding early in the day must spread today's rows across the elapsed part +// of the day instead of clamping them all to the same minute. +func TestGenerateSynthetic_TodayNotClamped(t *testing.T) { + now := time.Date(2026, 6, 1, 6, 0, 0, 0, time.UTC) + dayStart := time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC) + + seen := map[int64]bool{} + todayRows := 0 + for _, r := range GenerateSynthetic(now) { + reversedAt := time.UnixMilli(int64(r.ReversedAt)).UTC() + if reversedAt.Before(dayStart) { + continue + } + todayRows++ + seen[reversedAt.Truncate(time.Minute).Unix()] = true + } + if todayRows < 2 { + t.Skipf("only %d rows generated for today", todayRows) + } + if len(seen) < 2 { + t.Errorf("all %d of today's rows share one minute", todayRows) + } +} + +// Generating data with a clock that predates models.Epoch must not +// underflow the unsigned snowflake timestamp subtraction. Every decoded +// timestamp must be >= Epoch. +func TestGenerateSynthetic_NoEpochUnderflow(t *testing.T) { + // A clock one day before Epoch makes every generated createdAt precede + // Epoch, so the guard must clamp each snowflake timestamp to exactly + // Epoch. Without the guard the unsigned subtraction wraps to a huge + // value, decoding to a timestamp far beyond Epoch. + beforeEpoch := time.UnixMilli(int64(models.Epoch)).UTC().AddDate(0, 0, -1) + rows := GenerateSynthetic(beforeEpoch) + if len(rows) == 0 { + t.Fatal("expected synthetic rows") + } + for _, r := range rows { + ts := models.ParseSnowflake(r.ID).Timestamp + if ts != models.Epoch { + t.Fatalf("snowflake timestamp %d != Epoch %d (underflow not guarded)", ts, models.Epoch) + } + } +}