From cc7a4e47180e92ed0ea4eb37eed71522e3d0bd44 Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Thu, 20 Aug 2026 19:52:01 +0330 Subject: [PATCH] feat: stream and reconcile challenge dataset --- cmd/importer/main.go | 36 ++++++ internal/importer/importer.go | 200 +++++++++++++++++++++++++++++ internal/importer/importer_test.go | 44 +++++++ 3 files changed, 280 insertions(+) create mode 100644 cmd/importer/main.go create mode 100644 internal/importer/importer.go create mode 100644 internal/importer/importer_test.go diff --git a/cmd/importer/main.go b/cmd/importer/main.go new file mode 100644 index 0000000..b401564 --- /dev/null +++ b/cmd/importer/main.go @@ -0,0 +1,36 @@ +package main + +import ( + "context" + "log/slog" + "os" + "os/signal" + + "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/config" + "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/database" + "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/importer" +) + +func main() { + logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) + cfg, err := config.Load() + if err != nil { + logger.Error("configuration failed", "error", err) + os.Exit(1) + } + ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) + defer cancel() + pool, err := database.Open(ctx, cfg.DatabaseURL) + if err != nil { + logger.Error("database connection failed", "error", err) + os.Exit(1) + } + defer pool.Close() + result, err := importer.Run(ctx, pool, cfg.ImportPath) + if err != nil { + logger.Error("import failed", "error", err) + os.Exit(1) + } + logger.Info("import complete", "noop", result.IsNoop, "source_sha256", result.SourceSHA256, + "source_rows", result.SourceRows, "session_rows", result.SessionRows, "attempt_rows", result.AttemptRows) +} diff --git a/internal/importer/importer.go b/internal/importer/importer.go new file mode 100644 index 0000000..7fec309 --- /dev/null +++ b/internal/importer/importer.go @@ -0,0 +1,200 @@ +package importer + +import ( + "compress/gzip" + "context" + "crypto/sha256" + "encoding/csv" + "encoding/hex" + "errors" + "fmt" + "io" + "os" + "strings" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +var expectedHeader = []string{ + "session_key", "try_seq", "terminal_key", "merchant_key", "category_id", + "category_title", "amount", "adjusted_fee", "session_status", "try_status", + "switch_response_code", "psp_code", "issuer_bank_code", "payer_card_key", + "verify_type", "init_time_ms", "verify_time_ms", "created_at", "try_created_at", + "verified_at", "settled_at", "expire_in", +} + +type Result struct { + IsNoop bool + SourceSHA256 string + SourceRows int64 + SessionRows int64 + AttemptRows int64 +} + +func Run(ctx context.Context, pool *pgxpool.Pool, path string) (Result, error) { + checksum, err := checksumFile(path) + if err != nil { + return Result{}, err + } + if err := validateHeader(path); err != nil { + return Result{}, err + } + + var completed bool + err = pool.QueryRow(ctx, `SELECT EXISTS ( + SELECT 1 FROM import_runs WHERE source_sha256=$1 AND status='COMPLETED' + )`, checksum).Scan(&completed) + if err != nil { + return Result{}, fmt.Errorf("checking import manifest: %w", err) + } + if completed { + return Result{IsNoop: true, SourceSHA256: checksum}, nil + } + + tx, err := pool.BeginTx(ctx, pgx.TxOptions{}) + if err != nil { + return Result{}, fmt.Errorf("starting import transaction: %w", err) + } + defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }() + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(943271)`); err != nil { + return Result{}, fmt.Errorf("locking importer: %w", err) + } + + var runID int64 + err = tx.QueryRow(ctx, `INSERT INTO import_runs(source_sha256,source_path,status) + VALUES ($1,$2,'RUNNING') RETURNING id`, checksum, path).Scan(&runID) + if err != nil { + return Result{}, fmt.Errorf("creating import manifest: %w", err) + } + if err := createStaging(ctx, tx); err != nil { + return Result{}, err + } + + file, err := os.Open(path) + if err != nil { + return Result{}, fmt.Errorf("opening import archive: %w", err) + } + defer file.Close() + gzipReader, err := gzip.NewReader(file) + if err != nil { + return Result{}, fmt.Errorf("opening gzip stream: %w", err) + } + defer gzipReader.Close() + + copySQL := `COPY import_staging (` + strings.Join(expectedHeader, ",") + `) FROM STDIN WITH (FORMAT csv, HEADER true)` + copyResult, err := tx.Conn().PgConn().CopyFrom(ctx, gzipReader, copySQL) + if err != nil { + return Result{}, fmt.Errorf("copying archive into staging: %w", err) + } + result := Result{SourceSHA256: checksum, SourceRows: copyResult.RowsAffected()} + if err := promote(ctx, tx, runID, &result); err != nil { + return Result{}, err + } + if err := tx.Commit(ctx); err != nil { + return Result{}, fmt.Errorf("committing import: %w", err) + } + return result, nil +} + +func checksumFile(path string) (string, error) { + file, err := os.Open(path) + if err != nil { + return "", fmt.Errorf("opening source for checksum: %w", err) + } + defer file.Close() + hash := sha256.New() + if _, err := io.Copy(hash, file); err != nil { + return "", fmt.Errorf("checksumming source: %w", err) + } + return hex.EncodeToString(hash.Sum(nil)), nil +} + +func validateHeader(path string) error { + file, err := os.Open(path) + if err != nil { + return fmt.Errorf("opening source for preflight: %w", err) + } + defer file.Close() + gzipReader, err := gzip.NewReader(file) + if err != nil { + return fmt.Errorf("opening source gzip for preflight: %w", err) + } + defer gzipReader.Close() + header, err := csv.NewReader(gzipReader).Read() + if err != nil { + return fmt.Errorf("reading source header: %w", err) + } + if len(header) != len(expectedHeader) { + return fmt.Errorf("source header has %d columns, expected %d", len(header), len(expectedHeader)) + } + for i := range expectedHeader { + if header[i] != expectedHeader[i] { + return fmt.Errorf("source header column %d is %q, expected %q", i+1, header[i], expectedHeader[i]) + } + } + return nil +} + +func createStaging(ctx context.Context, tx pgx.Tx) error { + columns := make([]string, 0, len(expectedHeader)) + for _, column := range expectedHeader { + columns = append(columns, pgx.Identifier{column}.Sanitize()+" text") + } + statement := `CREATE TEMP TABLE import_staging (` + strings.Join(columns, ",") + `) ON COMMIT DROP` + if _, err := tx.Exec(ctx, statement); err != nil { + return fmt.Errorf("creating import staging table: %w", err) + } + return nil +} + +func promote(ctx context.Context, tx pgx.Tx, runID int64, result *Result) error { + queries := []string{ + `INSERT INTO merchants(source_key,category_id,category_title,display_name) + SELECT DISTINCT merchant_key,category_id::bigint,category_title,'کسب‌وکار '||merchant_key + FROM import_staging ON CONFLICT (source_key) DO NOTHING`, + `INSERT INTO terminals(merchant_id,source_key,verify_type) + SELECT DISTINCT m.id,s.terminal_key,s.verify_type FROM import_staging s + JOIN merchants m ON m.source_key=s.merchant_key + ON CONFLICT (merchant_id,source_key) DO NOTHING`, + `INSERT INTO payment_sessions(import_run_id,merchant_id,terminal_id,source_key,status,amount_rial,fee_amount_rial,payer_card_key,created_at,expires_at,verified_at,settled_at) + SELECT DISTINCT ON (m.id,s.session_key) $1,m.id,t.id,s.session_key,s.session_status,s.amount::bigint,s.adjusted_fee::bigint, + nullif(s.payer_card_key,''),s.created_at::timestamp AT TIME ZONE 'Asia/Tehran', + nullif(s.expire_in,'')::timestamp AT TIME ZONE 'Asia/Tehran',nullif(s.verified_at,'')::timestamp AT TIME ZONE 'Asia/Tehran', + nullif(s.settled_at,'')::timestamp AT TIME ZONE 'Asia/Tehran' + FROM import_staging s JOIN merchants m ON m.source_key=s.merchant_key + JOIN terminals t ON t.merchant_id=m.id AND t.source_key=s.terminal_key + ORDER BY m.id,s.session_key,s.try_seq::integer DESC + ON CONFLICT (merchant_id,source_key) DO NOTHING`, + `INSERT INTO payment_attempts(session_id,sequence,status,created_at,init_time_ms,verify_time_ms) + SELECT ps.id,s.try_seq::integer,s.try_status,nullif(s.try_created_at,'')::timestamp AT TIME ZONE 'Asia/Tehran', + nullif(s.init_time_ms,'')::integer,nullif(s.verify_time_ms,'')::integer + FROM import_staging s JOIN merchants m ON m.source_key=s.merchant_key + JOIN payment_sessions ps ON ps.merchant_id=m.id AND ps.source_key=s.session_key + WHERE s.try_seq::integer > 0 AND s.try_status <> 'NoAttempt' + ON CONFLICT (session_id,sequence) DO NOTHING`, + } + for i, query := range queries { + args := []any{} + if i == 2 { + args = append(args, runID) + } + if _, err := tx.Exec(ctx, query, args...); err != nil { + return fmt.Errorf("promoting import phase %d: %w", i+1, err) + } + } + if err := tx.QueryRow(ctx, `SELECT count(*) FROM payment_sessions WHERE import_run_id=$1`, runID).Scan(&result.SessionRows); err != nil { + return fmt.Errorf("counting imported sessions: %w", err) + } + if err := tx.QueryRow(ctx, `SELECT count(*) FROM payment_attempts pa JOIN payment_sessions ps ON ps.id=pa.session_id WHERE ps.import_run_id=$1`, runID).Scan(&result.AttemptRows); err != nil { + return fmt.Errorf("counting imported attempts: %w", err) + } + if result.SessionRows == 0 || result.AttemptRows > result.SourceRows { + return errors.New("import reconciliation failed") + } + _, err := tx.Exec(ctx, `UPDATE import_runs SET status='COMPLETED',source_rows=$2,session_rows=$3,attempt_rows=$4,completed_at=now() WHERE id=$1`, runID, result.SourceRows, result.SessionRows, result.AttemptRows) + if err != nil { + return fmt.Errorf("completing import manifest: %w", err) + } + return nil +} diff --git a/internal/importer/importer_test.go b/internal/importer/importer_test.go new file mode 100644 index 0000000..00398c5 --- /dev/null +++ b/internal/importer/importer_test.go @@ -0,0 +1,44 @@ +package importer + +import ( + "compress/gzip" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestValidateHeader(t *testing.T) { + t.Parallel() + tests := []struct { + name string + header string + wantErr bool + }{ + {name: "approved header", header: strings.Join(expectedHeader, ",")}, + {name: "unknown column", header: "wrong", wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "input.csv.gz") + file, err := os.Create(path) + if err != nil { + t.Fatal(err) + } + writer := gzip.NewWriter(file) + if _, err := writer.Write([]byte(tt.header + "\n")); err != nil { + t.Fatal(err) + } + if err := writer.Close(); err != nil { + t.Fatal(err) + } + if err := file.Close(); err != nil { + t.Fatal(err) + } + err = validateHeader(path) + if (err != nil) != tt.wantErr { + t.Fatalf("validateHeader() error = %v, wantErr %v", err, tt.wantErr) + } + }) + } +}