From 3f4802b24c45369cf3ee5e548e71dff6ac0b620f Mon Sep 17 00:00:00 2001 From: tiennm99 Date: Sun, 28 Jun 2026 09:58:28 +0700 Subject: [PATCH] feat(migrate): add DynamoDB-to-MongoDB data migrator Add standalone migrator tool with scan-and-transform logic for moving persisted state from DynamoDB to MongoDB. Includes tests and README with usage instructions. --- cmd/migrate-dynamo-to-mongo/README.md | 78 ++++++++ cmd/migrate-dynamo-to-mongo/main.go | 218 +++++++++++++++++++++++ cmd/migrate-dynamo-to-mongo/main_test.go | 156 ++++++++++++++++ 3 files changed, 452 insertions(+) create mode 100644 cmd/migrate-dynamo-to-mongo/README.md create mode 100644 cmd/migrate-dynamo-to-mongo/main.go create mode 100644 cmd/migrate-dynamo-to-mongo/main_test.go diff --git a/cmd/migrate-dynamo-to-mongo/README.md b/cmd/migrate-dynamo-to-mongo/README.md new file mode 100644 index 0000000..ee1d297 --- /dev/null +++ b/cmd/migrate-dynamo-to-mongo/README.md @@ -0,0 +1,78 @@ +# migrate-dynamo-to-mongo + +One-off CLI that copies every item from the prod DynamoDB KV table +(`miti99bot-data`) into MongoDB Atlas using the exact document schema the live +app writes, then verifies per-module parity. Idempotent and re-runnable. + +## What it does + +- Full-table `Scan` of DynamoDB (the table is a small KV) → group by `pk`. +- Writes each item through `storage.MongoKVStore.Put`, so the `value` encoding + is **byte-identical** to what the running bot writes (BSON binary). This is + what keeps re-runs idempotent and future CompareAndSwap correct — a raw + `UpdateOne` that stored `value` as a BSON string would silently break both. +- `Put` runs `validateKey` on each sort key and upserts by `_id`, so a re-run + produces no duplicates and an invalid key fails loud (no key the app's read + path can't load is ever written). + +| DynamoDB | MongoDB | +|---|---| +| `pk` (module name) | collection name | +| `sk` (user key) | document `_id` | +| `value` (JSON string) | `value` field (BSON binary, same bytes) | +| `updatedAt` | restamped to migration time (write-only field; nothing reads it) | + +## Usage + +```sh +export MONGO_URL='mongodb+srv://botuser:PASS@cluster0.xxxxx.mongodb.net/?retryWrites=true&w=majority' +export MONGO_DATABASE=miti99bot +export AWS_PROFILE=miti99bot-migrate # READ-ONLY profile (see IAM below) + +# 1. Dry run — report per-module counts, write nothing. +go run ./cmd/migrate-dynamo-to-mongo --dry-run +# or: make migrate-dynamo-to-mongo DRY_RUN=1 + +# 2. Real migration. +go run ./cmd/migrate-dynamo-to-mongo +# or: make migrate-dynamo-to-mongo + +# 3. Verify — per-module counts must match; exits non-zero on mismatch. +go run ./cmd/migrate-dynamo-to-mongo --verify +# or: make migrate-verify +``` + +Flags: `--dynamodb-table` (default `miti99bot-data`), `--dry-run`, `--verify`. + +For a local end-to-end test, point at DynamoDB Local + a local Mongo: + +```sh +DYNAMODB_LOCAL_URL=http://localhost:8001 \ +MONGODB_TEST_URL=mongodb://127.0.0.1:27017 \ +MONGO_DATABASE=migrate_test \ +go test ./cmd/migrate-dynamo-to-mongo/ -run TestMigrateAndVerify +``` + +## IAM — least privilege + +The runner needs **exactly** `dynamodb:Scan` on the table ARN and nothing else. +Verify uses a Scan tally (not Query), so no `dynamodb:Query` is needed; there +are **no write actions on the source**, enforcing the read-only requirement and +removing the destructive-credential foot-gun. + +```json +{ + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Allow", + "Action": "dynamodb:Scan", + "Resource": "arn:aws:dynamodb:ap-southeast-1:225603493174:table/miti99bot-data" + }] +} +``` + +## Cutover runbook + +The full zero-loss cutover (disable EventBridge → `deleteWebhook` → migrate → +verify → start the polling container) lives in +[`docs/deploy-coolify-selfhosted.md`](../../docs/deploy-coolify-selfhosted.md#cutover-runbook). diff --git a/cmd/migrate-dynamo-to-mongo/main.go b/cmd/migrate-dynamo-to-mongo/main.go new file mode 100644 index 0000000..f89a37d --- /dev/null +++ b/cmd/migrate-dynamo-to-mongo/main.go @@ -0,0 +1,218 @@ +// Command migrate-dynamo-to-mongo copies every item from the prod DynamoDB KV +// table into MongoDB Atlas using the same document schema the live app writes, +// then verifies per-module parity. It is idempotent (re-runs overwrite by key, +// never duplicate) and read-only on DynamoDB (Scan only). +// +// Usage: +// +// migrate-dynamo-to-mongo [--dynamodb-table miti99bot-data] [--dry-run] [--verify] +// +// Required env: MONGO_URL, MONGO_DATABASE. DynamoDB credentials come from the +// AWS default chain — use a DEDICATED READ-ONLY profile whose policy grants +// exactly `dynamodb:Scan` on the table ARN (nothing else, no write actions). +// For local testing, set DYNAMODB_LOCAL_URL to point at DynamoDB Local. +// +// - default: scan + write every item through MongoKVStore.Put (byte-identical +// value encoding to the app), then report per-module counts. +// - --dry-run: scan + report counts, write nothing. +// - --verify: tally DynamoDB per pk via Scan vs Mongo CountDocuments per +// collection; print a table and exit non-zero on any mismatch. +// +// Note: writing through Put restamps updatedAt to migration time. Values are +// byte-identical; nothing in the app reads updatedAt (write-only today), and +// --verify compares counts, so this is intentional and harmless. +package main + +import ( + "context" + "flag" + "fmt" + "os" + "sort" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/dynamodb" + "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + + "github.com/tiennm99/miti99bot/internal/storage" +) + +const defaultTable = "miti99bot-data" + +func main() { + table := flag.String("dynamodb-table", defaultTable, "source DynamoDB table name") + dryRun := flag.Bool("dry-run", false, "scan and report counts without writing") + verify := flag.Bool("verify", false, "compare per-module counts DynamoDB vs Mongo and exit non-zero on mismatch") + flag.Parse() + + if err := run(*table, *dryRun, *verify); err != nil { + fmt.Fprintln(os.Stderr, "error:", err) + os.Exit(1) + } +} + +func run(table string, dryRun, verify bool) error { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + + mongoURL := os.Getenv("MONGO_URL") + mongoDB := os.Getenv("MONGO_DATABASE") + if mongoURL == "" || mongoDB == "" { + return fmt.Errorf("MONGO_URL and MONGO_DATABASE are required") + } + + ddb, err := storage.NewDynamoDBClient(ctx, storage.DynamoDBEndpointFromEnv()) + if err != nil { + return fmt.Errorf("dynamodb client: %w", err) + } + + mclient, err := storage.NewMongoClient(ctx, mongoURL) + if err != nil { + return fmt.Errorf("mongo client: %w", err) + } + defer func() { _ = mclient.Disconnect(context.Background()) }() + mdb, err := storage.NewMongoDatabase(mclient, mongoDB) + if err != nil { + return err + } + + if verify { + return runVerify(ctx, ddb, mdb, table) + } + return runMigrate(ctx, ddb, mdb, table, dryRun) +} + +// item is one decoded DynamoDB KV row. +type item struct { + pk string // module name → collection + sk string // user key → _id + value []byte // raw value bytes +} + +// scanTable reads every row from the table via a full Scan (the KV table is +// small) and decodes pk/sk/value. Scan is the ONLY DynamoDB action used, so the +// runner needs just `dynamodb:Scan` on the table ARN. +func scanTable(ctx context.Context, ddb *dynamodb.Client, table string) ([]item, error) { + var items []item + pager := dynamodb.NewScanPaginator(ddb, &dynamodb.ScanInput{TableName: aws.String(table)}) + for pager.HasMorePages() { + page, err := pager.NextPage(ctx) + if err != nil { + return nil, fmt.Errorf("scan %s: %w", table, err) + } + for _, raw := range page.Items { + it, err := decodeItem(raw) + if err != nil { + return nil, err + } + items = append(items, it) + } + } + return items, nil +} + +func decodeItem(raw map[string]types.AttributeValue) (item, error) { + pk, ok := raw["pk"].(*types.AttributeValueMemberS) + if !ok { + return item{}, fmt.Errorf("item missing string pk: %v", raw) + } + sk, ok := raw["sk"].(*types.AttributeValueMemberS) + if !ok { + return item{}, fmt.Errorf("item %s missing string sk", pk.Value) + } + val, ok := raw["value"].(*types.AttributeValueMemberS) + if !ok { + return item{}, fmt.Errorf("item %s/%s missing string value", pk.Value, sk.Value) + } + return item{pk: pk.Value, sk: sk.Value, value: []byte(val.Value)}, nil +} + +// sortedKeys returns the map keys sorted for stable report output. +func sortedKeys(m map[string]int) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + sort.Strings(out) + return out +} + +// runMigrate scans the table, then (unless dry-run) writes every item through +// MongoKVStore.Put so the value encoding is byte-identical to what the live app +// writes — guaranteeing idempotent re-runs and correct future CAS. Put also +// runs validateKey on each sk and ReplaceOne-upserts by _id, so a re-run +// produces no duplicates and an invalid key fails loud rather than writing data +// the app's read path cannot load. +func runMigrate(ctx context.Context, ddb *dynamodb.Client, mdb *mongo.Database, table string, dryRun bool) error { + items, err := scanTable(ctx, ddb, table) + if err != nil { + return err + } + + counts := map[string]int{} + provider := storage.NewMongoProvider(mdb) + for _, it := range items { + counts[it.pk]++ + if dryRun { + continue + } + // For invalid module names For returns a store that errors on Put; + // Put validates the key. Either way an error aborts loudly. + if err := provider.For(it.pk).Put(ctx, it.sk, it.value); err != nil { + return fmt.Errorf("put %s/%s: %w", it.pk, it.sk, err) + } + } + + mode := "MIGRATED" + if dryRun { + mode = "DRY-RUN (no writes)" + } + fmt.Printf("%s — %d items across %d modules\n", mode, len(items), len(counts)) + for _, pk := range sortedKeys(counts) { + fmt.Printf(" %-20s %d\n", pk, counts[pk]) + } + return nil +} + +// runVerify tallies DynamoDB per pk via a Scan (so only dynamodb:Scan is +// needed — never Query) and compares against Mongo CountDocuments per +// collection. Prints a table and returns an error on any mismatch so the +// process exits non-zero. +func runVerify(ctx context.Context, ddb *dynamodb.Client, mdb *mongo.Database, table string) error { + items, err := scanTable(ctx, ddb, table) + if err != nil { + return err + } + ddbCounts := map[string]int{} + for _, it := range items { + ddbCounts[it.pk]++ + } + + mongoCounts := map[string]int{} + for pk := range ddbCounts { + n, err := mdb.Collection(pk).CountDocuments(ctx, bson.M{}) + if err != nil { + return fmt.Errorf("count mongo collection %s: %w", pk, err) + } + mongoCounts[pk] = int(n) + } + + fmt.Printf("%-20s %10s %10s %s\n", "MODULE", "DYNAMODB", "MONGO", "STATUS") + mismatch := false + for _, pk := range sortedKeys(ddbCounts) { + status := "OK" + if ddbCounts[pk] != mongoCounts[pk] { + status = "MISMATCH" + mismatch = true + } + fmt.Printf("%-20s %10d %10d %s\n", pk, ddbCounts[pk], mongoCounts[pk], status) + } + if mismatch { + return fmt.Errorf("verification failed: per-module counts differ") + } + fmt.Println("verification OK: all per-module counts match") + return nil +} diff --git a/cmd/migrate-dynamo-to-mongo/main_test.go b/cmd/migrate-dynamo-to-mongo/main_test.go new file mode 100644 index 0000000..7af13a2 --- /dev/null +++ b/cmd/migrate-dynamo-to-mongo/main_test.go @@ -0,0 +1,156 @@ +package main + +import ( + "context" + "os" + "reflect" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/dynamodb" + "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + "go.mongodb.org/mongo-driver/v2/mongo" + + "github.com/tiennm99/miti99bot/internal/storage" +) + +func TestDecodeItem(t *testing.T) { + raw := map[string]types.AttributeValue{ + "pk": &types.AttributeValueMemberS{Value: "coin"}, + "sk": &types.AttributeValueMemberS{Value: "user:1"}, + "value": &types.AttributeValueMemberS{Value: `{"x":1}`}, + } + it, err := decodeItem(raw) + if err != nil { + t.Fatalf("decodeItem: %v", err) + } + if it.pk != "coin" || it.sk != "user:1" || string(it.value) != `{"x":1}` { + t.Errorf("decoded %+v", it) + } + + // Missing value attribute is a hard error. + if _, err := decodeItem(map[string]types.AttributeValue{ + "pk": &types.AttributeValueMemberS{Value: "coin"}, + "sk": &types.AttributeValueMemberS{Value: "user:1"}, + }); err == nil { + t.Error("decodeItem with missing value: want error, got nil") + } +} + +func TestSortedKeys(t *testing.T) { + got := sortedKeys(map[string]int{"b": 1, "a": 2, "c": 3}) + if !reflect.DeepEqual(got, []string{"a", "b", "c"}) { + t.Errorf("sortedKeys = %v", got) + } +} + +// TestMigrateAndVerify is the end-to-end gate: seed DynamoDB Local across two +// modules, migrate into Mongo, assert values round-trip byte-identically, and +// confirm --verify reports matching counts. Skips unless BOTH emulators are +// configured. +func TestMigrateAndVerify(t *testing.T) { + ddbURL := os.Getenv("DYNAMODB_LOCAL_URL") + mongoURL := os.Getenv("MONGODB_TEST_URL") + mongoDB := os.Getenv("MONGO_DATABASE") + if ddbURL == "" || mongoURL == "" || mongoDB == "" { + t.Skip("set DYNAMODB_LOCAL_URL, MONGODB_TEST_URL, MONGO_DATABASE to run the migrator e2e test") + } + t.Setenv("AWS_ACCESS_KEY_ID", "test") + t.Setenv("AWS_SECRET_ACCESS_KEY", "test") + t.Setenv("AWS_REGION", "ap-southeast-1") + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + ddb, err := storage.NewDynamoDBClient(ctx, ddbURL) + if err != nil { + t.Fatalf("dynamodb client: %v", err) + } + table := "migrate-test" + createTable(t, ctx, ddb, table) + defer func() { + _, _ = ddb.DeleteTable(ctx, &dynamodb.DeleteTableInput{TableName: aws.String(table)}) + }() + + seed := []item{ + {pk: "coin", sk: "user:1", value: []byte(`{"bal":100}`)}, + {pk: "coin", sk: "user:2", value: []byte(`{"bal":200}`)}, + {pk: "stock", sk: "user:1", value: []byte(`{"vnd":5000}`)}, + } + for _, it := range seed { + putDynamoItem(t, ctx, ddb, table, it) + } + + // Migrate. + if err := runMigrate(ctx, ddb, mongoDatabase(t, ctx, mongoURL, mongoDB), table, false); err != nil { + t.Fatalf("runMigrate: %v", err) + } + // Re-run must stay idempotent. + if err := runMigrate(ctx, ddb, mongoDatabase(t, ctx, mongoURL, mongoDB), table, false); err != nil { + t.Fatalf("runMigrate re-run: %v", err) + } + // Verify passes. + if err := runVerify(ctx, ddb, mongoDatabase(t, ctx, mongoURL, mongoDB), table); err != nil { + t.Fatalf("runVerify: %v", err) + } + + // Spot-check byte-identical round trip through the app's read path. + db := mongoDatabase(t, ctx, mongoURL, mongoDB) + provider := storage.NewMongoProvider(db) + got, err := provider.For("coin").Get(ctx, "user:1") + if err != nil { + t.Fatalf("Get migrated value: %v", err) + } + if string(got) != `{"bal":100}` { + t.Errorf("migrated value = %q, want %q", got, `{"bal":100}`) + } +} + +func mongoDatabase(t *testing.T, ctx context.Context, uri, db string) *mongo.Database { + t.Helper() + client, err := storage.NewMongoClient(ctx, uri) + if err != nil { + t.Fatalf("NewMongoClient: %v", err) + } + t.Cleanup(func() { _ = client.Disconnect(context.Background()) }) + mdb, err := storage.NewMongoDatabase(client, db) + if err != nil { + t.Fatalf("NewMongoDatabase: %v", err) + } + return mdb +} + +func createTable(t *testing.T, ctx context.Context, c *dynamodb.Client, table string) { + t.Helper() + _, err := c.CreateTable(ctx, &dynamodb.CreateTableInput{ + TableName: aws.String(table), + BillingMode: types.BillingModePayPerRequest, + AttributeDefinitions: []types.AttributeDefinition{ + {AttributeName: aws.String("pk"), AttributeType: types.ScalarAttributeTypeS}, + {AttributeName: aws.String("sk"), AttributeType: types.ScalarAttributeTypeS}, + }, + KeySchema: []types.KeySchemaElement{ + {AttributeName: aws.String("pk"), KeyType: types.KeyTypeHash}, + {AttributeName: aws.String("sk"), KeyType: types.KeyTypeRange}, + }, + }) + if err != nil { + t.Fatalf("CreateTable: %v", err) + } +} + +func putDynamoItem(t *testing.T, ctx context.Context, c *dynamodb.Client, table string, it item) { + t.Helper() + _, err := c.PutItem(ctx, &dynamodb.PutItemInput{ + TableName: aws.String(table), + Item: map[string]types.AttributeValue{ + "pk": &types.AttributeValueMemberS{Value: it.pk}, + "sk": &types.AttributeValueMemberS{Value: it.sk}, + "value": &types.AttributeValueMemberS{Value: string(it.value)}, + }, + }) + if err != nil { + t.Fatalf("PutItem: %v", err) + } +}