package codes import ( "context" "database/sql" "fmt" "time" libcodes "github.com/wangjia/codes" dbx "github.com/wangjia/pangolin/server/internal/db" ) // PlanCode represents a plan tier. type PlanCode string const ( PlanFree PlanCode = "free" PlanPro PlanCode = "pro" PlanTeam PlanCode = "team" ) // planTier returns a numeric tier for comparison (higher = better). func planTier(p PlanCode) int { switch p { case PlanTeam: return 3 case PlanPro: return 2 default: return 1 } } // BatchChannel is the source channel for a code batch. type BatchChannel string const ( ChannelStore BatchChannel = "store" ChannelTG BatchChannel = "tg" ChannelLine BatchChannel = "line" ChannelManual BatchChannel = "manual" ) // SubscriptionRow mirrors the `subscriptions` DB row. type SubscriptionRow struct { ID int64 UserID int64 PlanID int64 PlanCode PlanCode ExpiresAt time.Time Source string } // Store wraps a *sql.DB and exposes all database operations needed by the // codes package. Every method that modifies data is safe to call inside or // outside an explicit transaction; methods that accept a *sql.Tx run within // that transaction. type Store struct { db *sql.DB dialect dbx.Dialect lib *libcodes.Store } // NewStore creates a Store backed by the given connection pool (MySQL or SQLite). func NewStore(db *sql.DB) *Store { d := dbx.DialectForDB(db) libD := libcodes.DialectMySQL if d.Name() == "sqlite" { libD = libcodes.DialectSQLite } return &Store{db: db, dialect: d, lib: libcodes.NewStore(db, libD)} } // Lib exposes the shared library store (used by Service / webhook shim). func (s *Store) Lib() *libcodes.Store { return s.lib } // -------------------------------------------------------------------------- // Subscription helpers (used inside redeem transaction) // -------------------------------------------------------------------------- // GetPlanID returns the primary key of plans.code. func (s *Store) GetPlanID(ctx context.Context, code PlanCode) (int64, error) { var id int64 err := s.db.QueryRowContext(ctx, `SELECT id FROM plans WHERE code=?`, string(code)).Scan(&id) if err != nil { return 0, fmt.Errorf("store.GetPlanID(%s): %w", code, err) } return id, nil } // GetPlanIDTx is GetPlanID inside a transaction (grant callback runs inside // the redeem tx and must not touch the pool). func (s *Store) GetPlanIDTx(ctx context.Context, tx *sql.Tx, code PlanCode) (int64, error) { var id int64 if err := tx.QueryRowContext(ctx, `SELECT id FROM plans WHERE code=?`, string(code)).Scan(&id); err != nil { return 0, fmt.Errorf("store.GetPlanIDTx(%s): %w", code, err) } return id, nil } // GetActiveSubscriptions returns all non-expired subscriptions for userID, // ordered by expires_at DESC. Called inside the redeem transaction. func (s *Store) GetActiveSubscriptions(ctx context.Context, tx *sql.Tx, userID int64) ([]SubscriptionRow, error) { rows, err := tx.QueryContext(ctx, `SELECT s.id, s.user_id, s.plan_id, p.code, s.expires_at, s.source FROM subscriptions s JOIN plans p ON p.id = s.plan_id WHERE s.user_id=? AND s.expires_at > ? ORDER BY s.expires_at DESC`, userID, time.Now().UTC()) if err != nil { return nil, fmt.Errorf("store.GetActiveSubscriptions: %w", err) } defer rows.Close() var subs []SubscriptionRow for rows.Next() { var sr SubscriptionRow if err := rows.Scan(&sr.ID, &sr.UserID, &sr.PlanID, &sr.PlanCode, &sr.ExpiresAt, &sr.Source); err != nil { return nil, fmt.Errorf("store.GetActiveSubscriptions scan: %w", err) } subs = append(subs, sr) } return subs, rows.Err() } // ExtendSubscription sets expires_at to max(current_expires_at, now) + duration. // Called inside the redeem transaction. The base date is computed in Go (the // subscription row is read inside the locked redeem tx, then updated) — portable // across engines and avoids DB-side DATE_ADD/GREATEST. func (s *Store) ExtendSubscription(ctx context.Context, tx *sql.Tx, subID int64, durationDays int) error { var current time.Time if err := tx.QueryRowContext(ctx, `SELECT expires_at FROM subscriptions WHERE id=?`, subID).Scan(¤t); err != nil { return fmt.Errorf("store.ExtendSubscription read: %w", err) } base := time.Now().UTC() if current.After(base) { base = current.UTC() } newExpiry := base.AddDate(0, 0, durationDays) _, err := tx.ExecContext(ctx, `UPDATE subscriptions SET expires_at = ? WHERE id=?`, newExpiry, subID) if err != nil { return fmt.Errorf("store.ExtendSubscription: %w", err) } return nil } // CreateSubscription inserts a new subscriptions row. expires_at is set to // max(now, latest_expires_at_for_same_plan) + duration_days. // latestSamePlan may be zero-value if the user has no existing sub for that plan. func (s *Store) CreateSubscription( ctx context.Context, tx *sql.Tx, userID, planID int64, durationDays int, latestSamePlan time.Time, source string, ) (int64, error) { now := time.Now().UTC() base := now if latestSamePlan.After(now) { base = latestSamePlan } expiresAt := base.AddDate(0, 0, durationDays) res, err := tx.ExecContext(ctx, `INSERT INTO subscriptions (user_id, plan_id, expires_at, source, created_at) VALUES (?, ?, ?, ?, ?)`, userID, planID, expiresAt, source, now) if err != nil { return 0, fmt.Errorf("store.CreateSubscription: %w", err) } id, _ := res.LastInsertId() return id, nil } // -------------------------------------------------------------------------- // Audit log // -------------------------------------------------------------------------- // WriteAuditLog inserts an audit_log row. actor and target are string // identifiers; meta is a JSON object (may be nil/empty). func (s *Store) WriteAuditLog(ctx context.Context, tx *sql.Tx, actor, action, target, metaJSON string) error { var err error if metaJSON == "" { metaJSON = "null" } now := time.Now().UTC() if tx != nil { _, err = tx.ExecContext(ctx, `INSERT INTO audit_log (actor, action, target, meta, at) VALUES (?, ?, ?, ?, ?)`, actor, action, target, metaJSON, now) } else { _, err = s.db.ExecContext(ctx, `INSERT INTO audit_log (actor, action, target, meta, at) VALUES (?, ?, ?, ?, ?)`, actor, action, target, metaJSON, now) } if err != nil { return fmt.Errorf("store.WriteAuditLog: %w", err) } return nil } // -------------------------------------------------------------------------- // Webhook ingestion (single externally-generated code per call) // -------------------------------------------------------------------------- // MintOne writes one externally-generated code (card-store webhook ingestion) // as one batch + one code row in a single transaction. A duplicate code_hash // returns ErrDuplicate with NOTHING committed — no orphan batch (the old // two-statement implementation leaked one batch row per duplicate delivery). func (s *Store) MintOne(ctx context.Context, codeHash string, ent libcodes.Entitlement, channel BatchChannel, createdBy, note string) error { tx, err := s.lib.BeginTx(ctx) if err != nil { return fmt.Errorf("store.MintOne: begin: %w", err) } defer tx.Rollback() batchID, err := s.lib.CreateBatchTx(ctx, tx, string(channel), ent, createdBy, note) if err != nil { return fmt.Errorf("store.MintOne: batch: %w", err) } if err := s.lib.CreateCodeTx(ctx, tx, codeHash, batchID, ent); err != nil { return err // ErrDuplicate passes through; defer 回滚保证无孤儿 } if err := tx.Commit(); err != nil { return fmt.Errorf("store.MintOne: commit: %w", err) } return nil } // BeginTx starts a new transaction at Read Committed isolation level. // The SELECT … FOR UPDATE in FindCodeByHashForUpdate provides the necessary // row-level exclusivity; Serializable is intentionally avoided to reduce // deadlock risk under concurrent redemptions. func (s *Store) BeginTx(ctx context.Context) (*sql.Tx, error) { return s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) }