mirror of
https://github.com/tiennm99/tiennm99bot.git
synced 2026-10-14 18:19:22 +00:00
chore: drop CF→AWS migration tooling and stale JS-port references
The CF→AWS data migration (closed 2026-05-16) is long done and the tooling isn't wired into any production path. Remove the one-shot binary, its support package, and the migration runbook. In live code, replace 'JS-parity' / 'same shape as JS' / 'cross-runtime KV migration' comments with the real, stable reason for each behavior (wire-format invariant, null-vs-zero distinction, CloudWatch alarm field name, etc.). 24 files touched across lolschedule, loldle, wordle, twentyq, trading, misc, util, server, metrics, ai, keylock. - delete cmd/migrate_cf_data/ - delete internal/migration/ - delete docs/cf-to-aws-migration-runbook.md
This commit is contained in:
tiennm99 committed
1 parent
c901f3ad40
commit
ec77a24db7
47 files changed
+142
-1491
No files matched your search
@@ -1,115 +0,0 @@
|
||||
// One-shot rewrite of the table's `value` attribute from Binary (legacy
|
||||
// shape) to String (current shape). Operator-elective; needed once after the
|
||||
// runtime swap from MemberB to MemberS in internal/storage/dynamodb_kv.go.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
awsconfig "github.com/aws/aws-sdk-go-v2/config"
|
||||
"github.com/aws/aws-sdk-go-v2/service/dynamodb"
|
||||
"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
|
||||
)
|
||||
|
||||
// signalContext is defined in main.go and gives every subcommand a SIGINT /
|
||||
// SIGTERM-cancellable context — Ctrl-C mid-scan now propagates as a clean
|
||||
// context error instead of leaving a half-converted table.
|
||||
//
|
||||
// (signature mirrors signal.NotifyContext for documentation purposes; no
|
||||
// re-declaration here, just a pointer for future readers.)
|
||||
|
||||
|
||||
func runConvertValueToString(args []string) error {
|
||||
fs := flag.NewFlagSet("convert-value-to-string", flag.ExitOnError)
|
||||
table := fs.String("table", "", "target DynamoDB table (required)")
|
||||
dryRun := fs.Bool("dry-run", false, "log actions but do not write")
|
||||
if err := fs.Parse(args); err != nil {
|
||||
return err
|
||||
}
|
||||
if *table == "" {
|
||||
return fmt.Errorf("--table is required")
|
||||
}
|
||||
|
||||
ctx, cancel := signalContext()
|
||||
defer cancel()
|
||||
cfg, err := awsconfig.LoadDefaultConfig(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("aws config: %w", err)
|
||||
}
|
||||
client := dynamodb.NewFromConfig(cfg)
|
||||
|
||||
converted, alreadyString, skipped, failed := 0, 0, 0, 0
|
||||
pager := dynamodb.NewScanPaginator(client, &dynamodb.ScanInput{TableName: aws.String(*table)})
|
||||
for pager.HasMorePages() {
|
||||
page, err := pager.NextPage(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("scan: %w", err)
|
||||
}
|
||||
for _, item := range page.Items {
|
||||
pk, sk, ok := itemPKSK(item)
|
||||
if !ok {
|
||||
skipped++
|
||||
continue
|
||||
}
|
||||
valAttr, ok := item["value"]
|
||||
if !ok {
|
||||
skipped++
|
||||
continue
|
||||
}
|
||||
binAttr, isBinary := valAttr.(*types.AttributeValueMemberB)
|
||||
if !isBinary {
|
||||
alreadyString++
|
||||
continue
|
||||
}
|
||||
if *dryRun {
|
||||
fmt.Printf(" DRY-RUN would convert pk=%s sk=%s len=%d\n", pk, sk, len(binAttr.Value))
|
||||
converted++
|
||||
continue
|
||||
}
|
||||
if err := putAsString(ctx, client, *table, pk, sk, binAttr.Value); err != nil {
|
||||
fmt.Fprintf(os.Stderr, " put %s/%s: %v\n", pk, sk, err)
|
||||
failed++
|
||||
continue
|
||||
}
|
||||
converted++
|
||||
}
|
||||
}
|
||||
|
||||
fmt.Printf("\nconvert-value-to-string report\n")
|
||||
fmt.Printf(" converted (B → S): %d\n", converted)
|
||||
fmt.Printf(" already String: %d\n", alreadyString)
|
||||
fmt.Printf(" skipped (no pk/sk/value): %d\n", skipped)
|
||||
fmt.Printf(" failed: %d\n", failed)
|
||||
return nil
|
||||
}
|
||||
|
||||
func itemPKSK(item map[string]types.AttributeValue) (string, string, bool) {
|
||||
pkAttr, ok := item["pk"].(*types.AttributeValueMemberS)
|
||||
if !ok {
|
||||
return "", "", false
|
||||
}
|
||||
skAttr, ok := item["sk"].(*types.AttributeValueMemberS)
|
||||
if !ok {
|
||||
return "", "", false
|
||||
}
|
||||
return pkAttr.Value, skAttr.Value, true
|
||||
}
|
||||
|
||||
func putAsString(ctx context.Context, client *dynamodb.Client, table, pk, sk string, val []byte) error {
|
||||
_, err := client.PutItem(ctx, &dynamodb.PutItemInput{
|
||||
TableName: aws.String(table),
|
||||
Item: map[string]types.AttributeValue{
|
||||
"pk": &types.AttributeValueMemberS{Value: pk},
|
||||
"sk": &types.AttributeValueMemberS{Value: sk},
|
||||
"value": &types.AttributeValueMemberS{Value: string(val)},
|
||||
"updatedAt": &types.AttributeValueMemberN{Value: strconv.FormatInt(time.Now().UTC().UnixNano(), 10)},
|
||||
},
|
||||
})
|
||||
return err
|
||||
}
|
||||
@@ -1,243 +0,0 @@
|
||||
// Command migrate_cf_data moves durable data from the legacy Cloudflare
|
||||
// KV/D1 stack into the live AWS DynamoDB table. Operator-invoked only.
|
||||
//
|
||||
// Subcommands:
|
||||
//
|
||||
// inventory Read CF KV keys, apply the Phase 01 policy, print
|
||||
// a classification report. No writes anywhere.
|
||||
//
|
||||
// kv-import Copy migrate-action KV keys into DynamoDB.
|
||||
// Idempotent by default (attribute_not_exists guard).
|
||||
// Flags: --table, --dry-run, --overwrite.
|
||||
//
|
||||
// trading-audit-dump Stream D1 `trading_trades` rows to a JSONL file.
|
||||
// Audit-only; not an import input.
|
||||
// Flags: --out (required).
|
||||
//
|
||||
// convert-value-to-string
|
||||
// One-shot rewrite of the table's `value` attribute
|
||||
// from Binary (legacy shape) to String (current
|
||||
// shape). Idempotent — items already stored as
|
||||
// String are skipped.
|
||||
// Flags: --table, --dry-run.
|
||||
//
|
||||
// Required env:
|
||||
//
|
||||
// CLOUDFLARE_API_TOKEN — read-scoped token for KV + D1
|
||||
// CLOUDFLARE_ACCOUNT_ID — production CF account
|
||||
// CF_KV_NAMESPACE_ID — production KV namespace
|
||||
// CF_D1_DATABASE_ID — production D1 database (only needed for
|
||||
// trading-audit-dump)
|
||||
// AWS_REGION (or standard AWS SDK env) — only needed for kv-import
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
|
||||
awsconfig "github.com/aws/aws-sdk-go-v2/config"
|
||||
"github.com/aws/aws-sdk-go-v2/service/dynamodb"
|
||||
"github.com/tiennm99/miti99bot/internal/migration"
|
||||
)
|
||||
|
||||
// signalContext returns a context cancelled on Ctrl-C / SIGTERM. Used by every
|
||||
// subcommand so a mid-scan abort leaves a clean error trail instead of a
|
||||
// half-converted table the operator has to reason about.
|
||||
func signalContext() (context.Context, context.CancelFunc) {
|
||||
return signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
}
|
||||
|
||||
func main() {
|
||||
if len(os.Args) < 2 {
|
||||
usage()
|
||||
os.Exit(2)
|
||||
}
|
||||
sub := os.Args[1]
|
||||
args := os.Args[2:]
|
||||
var err error
|
||||
switch sub {
|
||||
case "inventory":
|
||||
err = runInventory(args)
|
||||
case "kv-import":
|
||||
err = runKVImport(args)
|
||||
case "trading-audit-dump":
|
||||
err = runTradingAuditDump(args)
|
||||
case "convert-value-to-string":
|
||||
err = runConvertValueToString(args)
|
||||
case "-h", "--help", "help":
|
||||
usage()
|
||||
return
|
||||
default:
|
||||
usage()
|
||||
os.Exit(2)
|
||||
}
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "error:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
func usage() {
|
||||
fmt.Fprintln(os.Stderr, "usage: migrate_cf_data <inventory|kv-import|trading-audit-dump|convert-value-to-string> [flags]")
|
||||
}
|
||||
|
||||
func runInventory(args []string) error {
|
||||
fs := flag.NewFlagSet("inventory", flag.ExitOnError)
|
||||
if err := fs.Parse(args); err != nil {
|
||||
return err
|
||||
}
|
||||
kv, err := newKVClient()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := signalContext()
|
||||
defer cancel()
|
||||
keys, err := kv.ListKeys(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list keys: %w", err)
|
||||
}
|
||||
migrate := map[string]int{}
|
||||
skip := map[string]int{}
|
||||
for _, k := range keys {
|
||||
d := migration.Classify(k)
|
||||
if d.Action == migration.ActionMigrate {
|
||||
migrate[migration.PrefixOf(k)]++
|
||||
} else {
|
||||
skip[d.Reason]++
|
||||
}
|
||||
}
|
||||
fmt.Printf("Cloudflare KV namespace contains %d keys.\n\n", len(keys))
|
||||
fmt.Println("Migrate-action keys by prefix:")
|
||||
for p, n := range migrate {
|
||||
fmt.Printf(" %-30s %d\n", p, n)
|
||||
}
|
||||
fmt.Println("\nSkip-action keys by reason:")
|
||||
for r, n := range skip {
|
||||
fmt.Printf(" %-30s %d\n", r, n)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func runKVImport(args []string) error {
|
||||
fs := flag.NewFlagSet("kv-import", flag.ExitOnError)
|
||||
table := fs.String("table", "", "target DynamoDB table (required)")
|
||||
dryRun := fs.Bool("dry-run", false, "log actions but do not write")
|
||||
overwrite := fs.Bool("overwrite", false, "drop attribute_not_exists guard")
|
||||
if err := fs.Parse(args); err != nil {
|
||||
return err
|
||||
}
|
||||
if *table == "" {
|
||||
return fmt.Errorf("--table is required")
|
||||
}
|
||||
kv, err := newKVClient()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := signalContext()
|
||||
defer cancel()
|
||||
var writer *migration.DynamoDBWriter
|
||||
if !*dryRun {
|
||||
cfg, err := awsconfig.LoadDefaultConfig(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("aws config: %w", err)
|
||||
}
|
||||
writer = migration.NewDynamoDBWriter(dynamodb.NewFromConfig(cfg), *table, *overwrite)
|
||||
}
|
||||
keys, err := kv.ListKeys(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list keys: %w", err)
|
||||
}
|
||||
report := migration.NewReport()
|
||||
for _, k := range keys {
|
||||
d := migration.Classify(k)
|
||||
if d.Action != migration.ActionMigrate {
|
||||
report.AddSkippedPolicy(d.Reason)
|
||||
continue
|
||||
}
|
||||
val, err := kv.GetValue(ctx, k)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, " get %s: %v\n", k, err)
|
||||
report.AddFailed(migration.PrefixOf(k))
|
||||
continue
|
||||
}
|
||||
if *dryRun {
|
||||
fmt.Printf(" DRY-RUN would write pk=%s sk=%s len=%d\n", d.PK, d.SK, len(val))
|
||||
report.AddImported(migration.PrefixOf(k))
|
||||
continue
|
||||
}
|
||||
switch err := writer.Put(ctx, d.PK, d.SK, val); err {
|
||||
case nil:
|
||||
report.AddImported(migration.PrefixOf(k))
|
||||
case migration.ErrItemExists:
|
||||
report.AddSkippedExisting(migration.PrefixOf(k))
|
||||
default:
|
||||
fmt.Fprintf(os.Stderr, " put %s/%s: %v\n", d.PK, d.SK, err)
|
||||
report.AddFailed(migration.PrefixOf(k))
|
||||
}
|
||||
}
|
||||
report.Format(os.Stdout)
|
||||
return nil
|
||||
}
|
||||
|
||||
func runTradingAuditDump(args []string) error {
|
||||
fs := flag.NewFlagSet("trading-audit-dump", flag.ExitOnError)
|
||||
out := fs.String("out", "", "output JSONL file path (required)")
|
||||
if err := fs.Parse(args); err != nil {
|
||||
return err
|
||||
}
|
||||
if *out == "" {
|
||||
return fmt.Errorf("--out is required")
|
||||
}
|
||||
d1, err := newD1Client()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := signalContext()
|
||||
defer cancel()
|
||||
rows, err := d1.Query(ctx,
|
||||
"SELECT id, user_id, symbol, side, qty, price_vnd, ts FROM trading_trades ORDER BY id", nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("d1 query: %w", err)
|
||||
}
|
||||
f, err := os.Create(*out)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Surface Close's error: an audit JSONL that fails to flush is silently
|
||||
// truncated otherwise. Encode succeeding doesn't guarantee fsync — if
|
||||
// the final Close hits ENOSPC the operator must see it (this file is
|
||||
// evidence; a partial dump is worse than no dump).
|
||||
enc := json.NewEncoder(f)
|
||||
for _, r := range rows {
|
||||
if err := enc.Encode(r); err != nil {
|
||||
_ = f.Close()
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := f.Close(); err != nil {
|
||||
return fmt.Errorf("close audit dump: %w", err)
|
||||
}
|
||||
fmt.Printf("Wrote %d rows to %s\n", len(rows), *out)
|
||||
return nil
|
||||
}
|
||||
|
||||
func newKVClient() (*migration.CloudflareKVClient, error) {
|
||||
token, account, ns := os.Getenv("CLOUDFLARE_API_TOKEN"), os.Getenv("CLOUDFLARE_ACCOUNT_ID"), os.Getenv("CF_KV_NAMESPACE_ID")
|
||||
if token == "" || account == "" || ns == "" {
|
||||
return nil, fmt.Errorf("set CLOUDFLARE_API_TOKEN, CLOUDFLARE_ACCOUNT_ID, CF_KV_NAMESPACE_ID")
|
||||
}
|
||||
return migration.NewCloudflareKVClient(account, ns, token), nil
|
||||
}
|
||||
|
||||
func newD1Client() (*migration.CloudflareD1Client, error) {
|
||||
token, account, db := os.Getenv("CLOUDFLARE_API_TOKEN"), os.Getenv("CLOUDFLARE_ACCOUNT_ID"), os.Getenv("CF_D1_DATABASE_ID")
|
||||
if token == "" || account == "" || db == "" {
|
||||
return nil, fmt.Errorf("set CLOUDFLARE_API_TOKEN, CLOUDFLARE_ACCOUNT_ID, CF_D1_DATABASE_ID")
|
||||
}
|
||||
return migration.NewCloudflareD1Client(account, db, token), nil
|
||||
}
|
||||
Reference in new issue
Block a user