feat: stream and reconcile challenge dataset
This commit is contained in:
200
internal/importer/importer.go
Normal file
200
internal/importer/importer.go
Normal file
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user