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.
2026-07-30 02:22:15 +00:00 · 2026-06-28 09:58:28 +07:00
parent 5f8819ae46
commit 3f4802b24c
3 changed files with 452 additions and 0 deletions
@@ -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).
@@ -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
+}
@@ -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)
+	}
+}