831 lines
35 KiB
Go
831 lines
35 KiB
Go
package taskclaim
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"math"
|
|
"math/big"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"cmbuyer/admin/internal/deviceauth"
|
|
)
|
|
|
|
const writeTimeout = 2 * time.Second
|
|
|
|
type Store struct {
|
|
database *sql.DB
|
|
secret []byte
|
|
leaseTTL time.Duration
|
|
now func() time.Time
|
|
random io.Reader
|
|
randomMu sync.Mutex
|
|
writeGate chan struct{}
|
|
// The unexported linearization hooks let package tests coordinate real SQLite
|
|
// transactions at the first write. Production construction always leaves them nil.
|
|
beforeLinearization func()
|
|
afterLinearization func()
|
|
}
|
|
|
|
func NewStore(database *sql.DB, secret []byte, leaseTTL time.Duration) (*Store, error) {
|
|
if database == nil {
|
|
return nil, errors.New("task claim database is required")
|
|
}
|
|
if len(secret) != sha256.Size {
|
|
return nil, errors.New("task claim secret must be 32 bytes")
|
|
}
|
|
if leaseTTL <= 0 {
|
|
return nil, errors.New("task claim lease TTL must be positive")
|
|
}
|
|
if _, err := database.Exec("SELECT attempt_id, claim_nonce, claim_token_sha256 FROM purchase_attempt_claims LIMIT 1"); err != nil {
|
|
return nil, errors.New("task claim migration is not available")
|
|
}
|
|
store := &Store{
|
|
database: database, secret: append([]byte(nil), secret...), leaseTTL: leaseTTL,
|
|
now: time.Now, random: rand.Reader, writeGate: make(chan struct{}, 1),
|
|
}
|
|
if err := store.validateSecretIsolation(); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := store.validateStoredClaims(context.Background()); err != nil {
|
|
return nil, err
|
|
}
|
|
return store, nil
|
|
}
|
|
|
|
// validateSecretIsolation ensures the HMAC key cannot also authenticate a device. The session
|
|
// secret comparison is performed while parsing configuration, before either secret is discarded.
|
|
func (store *Store) validateSecretIsolation() error {
|
|
digest := sha256.Sum256(store.secret)
|
|
var count int
|
|
if err := store.database.QueryRow(`SELECT COUNT(*) FROM device_credentials WHERE token_sha256 = ?`, digest[:]).Scan(&count); err != nil {
|
|
return errors.New("validate task claim secret isolation")
|
|
}
|
|
if count != 0 {
|
|
return errors.New("task claim secret must be isolated from device credentials")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// validateStoredClaims covers open and closed claims. Replacing the secret must fail startup;
|
|
// silently signing a new token would destroy idempotent recovery and the ownership audit chain.
|
|
func (store *Store) validateStoredClaims(ctx context.Context) error {
|
|
rows, err := store.database.QueryContext(ctx, `SELECT claims.claimed_by_device_id, claims.task_id, claims.authorization_id,
|
|
claims.attempt_id, claims.claim_generation, claims.claim_nonce, typeof(claims.claim_nonce), length(claims.claim_nonce),
|
|
claims.claim_token_sha256, typeof(claims.claim_token_sha256), length(claims.claim_token_sha256),
|
|
claims.authorization_task_version, claims.goods_id, claims.sku_color, claims.sku_size,
|
|
claims.quantity, claims.total_price_cap, claims.authorization_expires_at, claims.closed_at,
|
|
attempts.claim_generation, attempts.status, authorizations.status, tasks.status
|
|
FROM purchase_attempt_claims AS claims
|
|
LEFT JOIN purchase_attempts AS attempts ON attempts.id = claims.attempt_id
|
|
LEFT JOIN order_authorizations AS authorizations ON authorizations.id = claims.authorization_id
|
|
LEFT JOIN tasks ON tasks.id = claims.task_id
|
|
ORDER BY claims.attempt_id`)
|
|
if err != nil {
|
|
return errors.New("validate stored task claims")
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var deviceID, taskID, authorizationID, attemptID string
|
|
var generation, authorizationTaskVersion, quantity int
|
|
var nonce, storedHash []byte
|
|
var nonceType, hashType, goodsID, color, size, price, expires string
|
|
var nonceLength, hashLength int
|
|
var closed, attemptStatus, authorizationStatus, taskStatus sql.NullString
|
|
var attemptGeneration sql.NullInt64
|
|
if err := rows.Scan(&deviceID, &taskID, &authorizationID, &attemptID, &generation,
|
|
&nonce, &nonceType, &nonceLength, &storedHash, &hashType, &hashLength,
|
|
&authorizationTaskVersion, &goodsID, &color, &size, &quantity, &price, &expires, &closed,
|
|
&attemptGeneration, &attemptStatus, &authorizationStatus, &taskStatus); err != nil {
|
|
return errors.New("validate stored task claims")
|
|
}
|
|
if !deviceauth.ValidDeviceID(deviceID) || !validUUID(taskID) || !validUUID(authorizationID) || !validUUID(attemptID) ||
|
|
generation <= 0 || nonceType != "blob" || nonceLength != sha256.Size || len(nonce) != sha256.Size ||
|
|
hashType != "blob" || hashLength != sha256.Size || len(storedHash) != sha256.Size ||
|
|
authorizationTaskVersion <= 0 || !digitsOnly(goodsID) || color == "" || size == "" || quantity <= 0 ||
|
|
!canonicalMoney(price) || !validCanonicalTime(expires) || (closed.Valid && !validCanonicalTime(closed.String)) ||
|
|
!attemptGeneration.Valid || attemptGeneration.Int64 != int64(generation) ||
|
|
!validAttemptStatus(attemptStatus) || !validAuthorizationStatus(authorizationStatus) || !validTaskStatus(taskStatus) {
|
|
return errors.New("stored task claim metadata is invalid")
|
|
}
|
|
token := deriveToken(store.secret, deviceID, taskID, authorizationID, attemptID, generation, nonce)
|
|
if !matchingHash(tokenHash(token), storedHash) {
|
|
return errors.New("task claim secret does not match stored claims")
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return errors.New("validate stored task claims")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (store *Store) ClaimNext(ctx context.Context, deviceID string, command ClaimCommand) (ClaimResponse, bool, error) {
|
|
if !deviceauth.ValidDeviceID(deviceID) || !validUUID(command.SessionID) || !validUUID(command.ClaimRequestID) {
|
|
return ClaimResponse{}, false, ErrInvalid
|
|
}
|
|
writeCtx, cancel := context.WithTimeout(ctx, writeTimeout)
|
|
defer cancel()
|
|
select {
|
|
case store.writeGate <- struct{}{}:
|
|
defer func() { <-store.writeGate }()
|
|
case <-writeCtx.Done():
|
|
return ClaimResponse{}, false, writeCtx.Err()
|
|
}
|
|
|
|
transaction, err := store.database.BeginTx(writeCtx, nil)
|
|
if err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
defer transaction.Rollback()
|
|
|
|
// This must be the transaction's first database statement. The no-op conditional UPDATE takes
|
|
// SQLite's write position and linearizes a concurrent credential revocation before any replay,
|
|
// EMPTY response, conflict response, candidate read, or other business write is possible.
|
|
if store.beforeLinearization != nil {
|
|
store.beforeLinearization()
|
|
}
|
|
active, err := transaction.ExecContext(writeCtx, `UPDATE device_credentials SET status = status
|
|
WHERE device_id = ? AND status = 'ACTIVE' AND revoked_at IS NULL`, deviceID)
|
|
if err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
if ok, err := exactlyOne(active); err != nil {
|
|
return ClaimResponse{}, false, err
|
|
} else if !ok {
|
|
return ClaimResponse{}, false, ErrDeviceInactive
|
|
}
|
|
if store.afterLinearization != nil {
|
|
store.afterLinearization()
|
|
}
|
|
|
|
now, err := store.serverNow()
|
|
if err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
request, found, err := findClaimRequest(writeCtx, transaction, command.ClaimRequestID)
|
|
if err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
if found {
|
|
if request.DeviceID != deviceID || request.SessionID != command.SessionID {
|
|
return ClaimResponse{}, false, ErrIdempotencyConflict
|
|
}
|
|
switch request.Outcome {
|
|
case "EMPTY":
|
|
if err := transaction.Commit(); err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
return ClaimResponse{}, false, nil
|
|
case "BLOCKED":
|
|
if err := transaction.Commit(); err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
return ClaimResponse{}, false, ErrRequiresManual
|
|
case "CLAIMED":
|
|
record, found, err := store.loadClaimByAttempt(writeCtx, transaction, request.AttemptID)
|
|
if err != nil || !found {
|
|
if err == nil {
|
|
err = errors.New("stored claim request has no claim")
|
|
}
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
response, err := store.responseFor(record, request.ResponseLeaseExpiresAt)
|
|
if err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
if err := transaction.Commit(); err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
return response, true, nil
|
|
default:
|
|
return ClaimResponse{}, false, errors.New("stored claim request outcome is invalid")
|
|
}
|
|
}
|
|
|
|
existing, found, err := store.loadOpenClaimByDevice(writeCtx, transaction, deviceID)
|
|
if err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
if found {
|
|
current := existing.SessionID == command.SessionID && existing.ClosedAt == "" &&
|
|
existing.LeaseExpiresAt.After(now) && existing.AuthorizationExpiresAt.After(now) &&
|
|
existing.CurrentAuthorizationExpiresAt.After(now) && existing.AuthorizationStatus == "CLAIMED" &&
|
|
existing.authorizationConsistent() && existing.recoverableBusinessState()
|
|
if !current {
|
|
if err := insertClaimRequest(writeCtx, transaction, command.ClaimRequestID, deviceID, command.SessionID, "BLOCKED", "", "", "manual_recovery_required", now); err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
if err := transaction.Commit(); err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
return ClaimResponse{}, false, ErrRequiresManual
|
|
}
|
|
response, err := store.responseFor(existing, existing.LeaseExpiresText)
|
|
if err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
if err := insertClaimRequest(writeCtx, transaction, command.ClaimRequestID, deviceID, command.SessionID, "CLAIMED", existing.AttemptID, existing.LeaseExpiresText, "", now); err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
if err := transaction.Commit(); err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
return response, true, nil
|
|
}
|
|
|
|
candidate, found, err := findCandidate(writeCtx, transaction, now)
|
|
if err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
if !found {
|
|
if err := insertClaimRequest(writeCtx, transaction, command.ClaimRequestID, deviceID, command.SessionID, "EMPTY", "", "", "", now); err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
if err := transaction.Commit(); err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
return ClaimResponse{}, false, nil
|
|
}
|
|
|
|
generation, err := nextGeneration(writeCtx, transaction, candidate.TaskID)
|
|
if err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
attemptID, err := store.newUUID()
|
|
if err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
nonce, err := store.randomBytes(sha256.Size)
|
|
if err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
token := deriveToken(store.secret, deviceID, candidate.TaskID, candidate.AuthorizationID, attemptID, generation, nonce)
|
|
storedTokenHash := tokenHash(token)
|
|
leaseExpires := now.Add(store.leaseTTL)
|
|
if candidate.AuthorizationExpiresAt.Before(leaseExpires) {
|
|
leaseExpires = candidate.AuthorizationExpiresAt
|
|
}
|
|
leaseText := formatTime(leaseExpires)
|
|
nowText := formatTime(now)
|
|
|
|
authorizationUpdate, err := transaction.ExecContext(writeCtx, `UPDATE order_authorizations SET status = 'CLAIMED'
|
|
WHERE id = ? AND task_id = ? AND status = 'ACTIVE' AND task_version = ?
|
|
AND goods_id = ? AND sku_color = ? AND sku_size = ? AND quantity = ?
|
|
AND total_price_cap = ? AND expires_at = ?`,
|
|
candidate.AuthorizationID, candidate.TaskID, candidate.TaskVersion, candidate.GoodsID,
|
|
candidate.SKUColor, candidate.SKUSize, candidate.Quantity, candidate.TotalPriceCap,
|
|
candidate.AuthorizationExpiresText)
|
|
if err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
if ok, err := exactlyOne(authorizationUpdate); err != nil || !ok {
|
|
if err == nil {
|
|
err = errors.New("authorization changed during claim")
|
|
}
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
taskUpdate, err := transaction.ExecContext(writeCtx, `UPDATE tasks SET status = 'CLAIMED', version = version + 1, updated_at = ?
|
|
WHERE id = ? AND status = 'PENDING' AND version = ? AND title = ? AND goods_id = ?
|
|
AND sku_color = ? AND sku_size = ? AND quantity = ? AND max_total_price = ?`,
|
|
nowText, candidate.TaskID, candidate.TaskVersion, candidate.Title, candidate.GoodsID,
|
|
candidate.SKUColor, candidate.SKUSize, candidate.Quantity, candidate.TotalPriceCap)
|
|
if err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
if ok, err := exactlyOne(taskUpdate); err != nil || !ok {
|
|
if err == nil {
|
|
err = errors.New("task changed during claim")
|
|
}
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
if _, err := transaction.ExecContext(writeCtx, `INSERT INTO purchase_attempts
|
|
(id, task_id, authorization_id, claim_generation, status, started_at)
|
|
VALUES (?, ?, ?, ?, 'CLAIMED', ?)`, attemptID, candidate.TaskID, candidate.AuthorizationID, generation, nowText); err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
if _, err := transaction.ExecContext(writeCtx, `INSERT INTO purchase_attempt_claims
|
|
(attempt_id, task_id, authorization_id, claimed_by_device_id, session_id, claim_generation,
|
|
task_version, task_title, authorization_task_version, goods_id, sku_color, sku_size, quantity,
|
|
total_price_cap, authorization_expires_at, claim_nonce, claim_token_sha256,
|
|
lease_expires_at, claimed_at, closed_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL)`,
|
|
attemptID, candidate.TaskID, candidate.AuthorizationID, deviceID, command.SessionID, generation,
|
|
candidate.TaskVersion+1, candidate.Title, candidate.TaskVersion, candidate.GoodsID,
|
|
candidate.SKUColor, candidate.SKUSize, candidate.Quantity, candidate.TotalPriceCap,
|
|
candidate.AuthorizationExpiresText, nonce, storedTokenHash, leaseText, nowText); err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
if err := insertClaimRequest(writeCtx, transaction, command.ClaimRequestID, deviceID, command.SessionID, "CLAIMED", attemptID, leaseText, "", now); err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
response := ClaimResponse{
|
|
Task: ClaimedTask{ID: candidate.TaskID, Version: candidate.TaskVersion + 1, Title: candidate.Title,
|
|
ProductURL: productURL(candidate.GoodsID), GoodsID: candidate.GoodsID, SKUColor: candidate.SKUColor,
|
|
SKUSize: candidate.SKUSize, Quantity: candidate.Quantity, MaxTotalPrice: candidate.TotalPriceCap},
|
|
Authorization: ClaimedAuthorization{ID: candidate.AuthorizationID, TaskVersion: candidate.TaskVersion, ExpiresAt: candidate.AuthorizationExpiresText},
|
|
Attempt: ClaimedAttempt{ID: attemptID, ClaimToken: hex.EncodeToString(token), ClaimGeneration: generation, LeaseExpiresAt: leaseText},
|
|
}
|
|
if err := transaction.Commit(); err != nil {
|
|
return ClaimResponse{}, false, err
|
|
}
|
|
return response, true, nil
|
|
}
|
|
|
|
func (store *Store) Renew(ctx context.Context, deviceID string, command RenewCommand) (RenewResponse, error) {
|
|
providedToken, tokenOK := decodeToken(command.ClaimToken)
|
|
if !deviceauth.ValidDeviceID(deviceID) || !validUUID(command.TaskID) || !validUUID(command.RenewRequestID) ||
|
|
!validUUID(command.SessionID) || !validUUID(command.AttemptID) || command.ClaimGeneration <= 0 ||
|
|
!tokenOK || !validCanonicalTime(command.ExpectedLeaseExpiresAt) {
|
|
return RenewResponse{}, ErrInvalid
|
|
}
|
|
providedHash := tokenHash(providedToken)
|
|
writeCtx, cancel := context.WithTimeout(ctx, writeTimeout)
|
|
defer cancel()
|
|
select {
|
|
case store.writeGate <- struct{}{}:
|
|
defer func() { <-store.writeGate }()
|
|
case <-writeCtx.Done():
|
|
return RenewResponse{}, writeCtx.Err()
|
|
}
|
|
transaction, err := store.database.BeginTx(writeCtx, nil)
|
|
if err != nil {
|
|
return RenewResponse{}, err
|
|
}
|
|
defer transaction.Rollback()
|
|
|
|
// As in ClaimNext, this is deliberately the first database statement in the transaction.
|
|
if store.beforeLinearization != nil {
|
|
store.beforeLinearization()
|
|
}
|
|
active, err := transaction.ExecContext(writeCtx, `UPDATE device_credentials SET status = status
|
|
WHERE device_id = ? AND status = 'ACTIVE' AND revoked_at IS NULL`, deviceID)
|
|
if err != nil {
|
|
return RenewResponse{}, err
|
|
}
|
|
if ok, err := exactlyOne(active); err != nil {
|
|
return RenewResponse{}, err
|
|
} else if !ok {
|
|
return RenewResponse{}, ErrDeviceInactive
|
|
}
|
|
if store.afterLinearization != nil {
|
|
store.afterLinearization()
|
|
}
|
|
|
|
renewal, found, err := findRenewal(writeCtx, transaction, command.RenewRequestID)
|
|
if err != nil {
|
|
return RenewResponse{}, err
|
|
}
|
|
if found {
|
|
if renewal.TaskID != command.TaskID || renewal.AttemptID != command.AttemptID || renewal.DeviceID != deviceID ||
|
|
renewal.SessionID != command.SessionID || renewal.Generation != command.ClaimGeneration ||
|
|
renewal.ExpectedLeaseExpiresAt != command.ExpectedLeaseExpiresAt || !matchingHash(renewal.TokenHash, providedHash) {
|
|
return RenewResponse{}, ErrIdempotencyConflict
|
|
}
|
|
response := RenewResponse{TaskID: renewal.TaskID, AttemptID: renewal.AttemptID, ClaimGeneration: renewal.Generation, LeaseExpiresAt: renewal.LeaseExpiresAt}
|
|
if err := transaction.Commit(); err != nil {
|
|
return RenewResponse{}, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
now, err := store.serverNow()
|
|
if err != nil {
|
|
return RenewResponse{}, err
|
|
}
|
|
record, found, err := store.loadClaimByAttempt(writeCtx, transaction, command.AttemptID)
|
|
if err != nil {
|
|
return RenewResponse{}, err
|
|
}
|
|
if !found || record.TaskID != command.TaskID || record.DeviceID != deviceID || record.SessionID != command.SessionID ||
|
|
record.Generation != command.ClaimGeneration || !matchingHash(record.TokenHash, providedHash) {
|
|
return RenewResponse{}, ErrNotCurrent
|
|
}
|
|
stateCurrent := record.ClosedAt == "" && record.LeaseExpiresAt.After(now) && record.AuthorizationExpiresAt.After(now) &&
|
|
record.CurrentAuthorizationExpiresAt.After(now) && record.AuthorizationStatus == "CLAIMED" &&
|
|
record.authorizationConsistent() && record.recoverableBusinessState()
|
|
if !stateCurrent || record.LeaseExpiresText != command.ExpectedLeaseExpiresAt {
|
|
return RenewResponse{}, ErrNotCurrent
|
|
}
|
|
leaseExpires := now.Add(store.leaseTTL)
|
|
if record.AuthorizationExpiresAt.Before(leaseExpires) {
|
|
leaseExpires = record.AuthorizationExpiresAt
|
|
}
|
|
leaseText := formatTime(leaseExpires)
|
|
updated, err := transaction.ExecContext(writeCtx, `UPDATE purchase_attempt_claims SET lease_expires_at = ?
|
|
WHERE attempt_id = ? AND lease_expires_at = ? AND closed_at IS NULL`, leaseText, command.AttemptID, command.ExpectedLeaseExpiresAt)
|
|
if err != nil {
|
|
return RenewResponse{}, err
|
|
}
|
|
if ok, err := exactlyOne(updated); err != nil || !ok {
|
|
if err == nil {
|
|
err = ErrNotCurrent
|
|
}
|
|
return RenewResponse{}, err
|
|
}
|
|
if _, err := transaction.ExecContext(writeCtx, `INSERT INTO purchase_attempt_lease_renewals
|
|
(renew_request_id, task_id, attempt_id, device_id, session_id, claim_generation,
|
|
claim_token_sha256, expected_lease_expires_at, lease_expires_at, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
|
command.RenewRequestID, command.TaskID, command.AttemptID, deviceID, command.SessionID,
|
|
command.ClaimGeneration, record.TokenHash, command.ExpectedLeaseExpiresAt, leaseText, formatTime(now)); err != nil {
|
|
return RenewResponse{}, err
|
|
}
|
|
response := RenewResponse{TaskID: command.TaskID, AttemptID: command.AttemptID, ClaimGeneration: command.ClaimGeneration, LeaseExpiresAt: leaseText}
|
|
if err := transaction.Commit(); err != nil {
|
|
return RenewResponse{}, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
type claimRequestRecord struct {
|
|
DeviceID, SessionID, Outcome, AttemptID, ResponseLeaseExpiresAt string
|
|
}
|
|
|
|
func findClaimRequest(ctx context.Context, transaction *sql.Tx, requestID string) (claimRequestRecord, bool, error) {
|
|
var record claimRequestRecord
|
|
var attemptID, responseLease sql.NullString
|
|
err := transaction.QueryRowContext(ctx, `SELECT device_id, session_id, outcome, attempt_id, response_lease_expires_at
|
|
FROM task_claim_requests WHERE claim_request_id = ?`, requestID).
|
|
Scan(&record.DeviceID, &record.SessionID, &record.Outcome, &attemptID, &responseLease)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return claimRequestRecord{}, false, nil
|
|
}
|
|
if err != nil {
|
|
return claimRequestRecord{}, false, err
|
|
}
|
|
record.AttemptID, record.ResponseLeaseExpiresAt = attemptID.String, responseLease.String
|
|
return record, true, nil
|
|
}
|
|
|
|
func insertClaimRequest(ctx context.Context, transaction *sql.Tx, requestID, deviceID, sessionID, outcome, attemptID, responseLease, errorCode string, now time.Time) error {
|
|
var attempt, lease, code any
|
|
if attemptID != "" {
|
|
attempt = attemptID
|
|
}
|
|
if responseLease != "" {
|
|
lease = responseLease
|
|
}
|
|
if errorCode != "" {
|
|
code = errorCode
|
|
}
|
|
_, err := transaction.ExecContext(ctx, `INSERT INTO task_claim_requests
|
|
(claim_request_id, device_id, session_id, outcome, attempt_id, response_lease_expires_at, error_code, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, requestID, deviceID, sessionID, outcome, attempt, lease, code, formatTime(now))
|
|
return err
|
|
}
|
|
|
|
type renewalRecord struct {
|
|
TaskID, AttemptID, DeviceID, SessionID string
|
|
Generation int
|
|
TokenHash []byte
|
|
ExpectedLeaseExpiresAt, LeaseExpiresAt string
|
|
}
|
|
|
|
func findRenewal(ctx context.Context, transaction *sql.Tx, requestID string) (renewalRecord, bool, error) {
|
|
var record renewalRecord
|
|
err := transaction.QueryRowContext(ctx, `SELECT task_id, attempt_id, device_id, session_id,
|
|
claim_generation, claim_token_sha256, expected_lease_expires_at, lease_expires_at
|
|
FROM purchase_attempt_lease_renewals WHERE renew_request_id = ?`, requestID).
|
|
Scan(&record.TaskID, &record.AttemptID, &record.DeviceID, &record.SessionID, &record.Generation,
|
|
&record.TokenHash, &record.ExpectedLeaseExpiresAt, &record.LeaseExpiresAt)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return renewalRecord{}, false, nil
|
|
}
|
|
return record, err == nil, err
|
|
}
|
|
|
|
type claimRecord struct {
|
|
AttemptID, TaskID, AuthorizationID, DeviceID, SessionID string
|
|
Generation, TaskVersion, CurrentTaskVersion int
|
|
TaskTitle string
|
|
Nonce, TokenHash []byte
|
|
LeaseExpiresText, ClaimedAt, ClosedAt string
|
|
LeaseExpiresAt time.Time
|
|
AuthorizationTaskVersion int
|
|
GoodsID, SKUColor, SKUSize, TotalPriceCap string
|
|
Quantity int
|
|
AuthorizationExpiresText, AuthorizationStatus string
|
|
AuthorizationExpiresAt time.Time
|
|
CurrentAuthorizationTaskVersion int
|
|
CurrentGoodsID, CurrentSKUColor, CurrentSKUSize string
|
|
CurrentQuantity int
|
|
CurrentTotalPriceCap, CurrentAuthorizationExpiresText string
|
|
CurrentAuthorizationExpiresAt time.Time
|
|
AttemptStatus, TaskStatus string
|
|
CurrentTaskTitle, CurrentTaskGoodsID string
|
|
CurrentTaskSKUColor, CurrentTaskSKUSize string
|
|
CurrentTaskQuantity int
|
|
CurrentTaskMaxTotalPrice string
|
|
CurrentAttemptGeneration int
|
|
}
|
|
|
|
const claimSelect = `SELECT claims.attempt_id, claims.task_id, claims.authorization_id,
|
|
claims.claimed_by_device_id, claims.session_id, claims.claim_generation, claims.task_version,
|
|
claims.task_title, claims.authorization_task_version, claims.goods_id, claims.sku_color,
|
|
claims.sku_size, claims.quantity, claims.total_price_cap, claims.authorization_expires_at,
|
|
claims.claim_nonce, claims.claim_token_sha256, claims.lease_expires_at,
|
|
claims.claimed_at, claims.closed_at, authorizations.task_version, authorizations.goods_id,
|
|
authorizations.sku_color, authorizations.sku_size, authorizations.quantity,
|
|
authorizations.total_price_cap, authorizations.expires_at, authorizations.status,
|
|
attempts.claim_generation, attempts.status, tasks.status, tasks.version, tasks.title, tasks.goods_id,
|
|
tasks.sku_color, tasks.sku_size, tasks.quantity, tasks.max_total_price
|
|
FROM purchase_attempt_claims AS claims
|
|
JOIN order_authorizations AS authorizations
|
|
ON authorizations.task_id = claims.task_id AND authorizations.id = claims.authorization_id
|
|
JOIN purchase_attempts AS attempts ON attempts.id = claims.attempt_id
|
|
JOIN tasks ON tasks.id = claims.task_id `
|
|
|
|
func (store *Store) loadOpenClaimByDevice(ctx context.Context, transaction *sql.Tx, deviceID string) (claimRecord, bool, error) {
|
|
return store.scanClaim(transaction.QueryRowContext(ctx, claimSelect+`WHERE claims.claimed_by_device_id = ? AND claims.closed_at IS NULL`, deviceID))
|
|
}
|
|
|
|
func (store *Store) loadClaimByAttempt(ctx context.Context, transaction *sql.Tx, attemptID string) (claimRecord, bool, error) {
|
|
return store.scanClaim(transaction.QueryRowContext(ctx, claimSelect+`WHERE claims.attempt_id = ?`, attemptID))
|
|
}
|
|
|
|
type rowScanner interface{ Scan(...any) error }
|
|
|
|
func (store *Store) scanClaim(row rowScanner) (claimRecord, bool, error) {
|
|
var record claimRecord
|
|
var closed sql.NullString
|
|
err := row.Scan(&record.AttemptID, &record.TaskID, &record.AuthorizationID, &record.DeviceID,
|
|
&record.SessionID, &record.Generation, &record.TaskVersion, &record.TaskTitle,
|
|
&record.AuthorizationTaskVersion, &record.GoodsID, &record.SKUColor, &record.SKUSize,
|
|
&record.Quantity, &record.TotalPriceCap, &record.AuthorizationExpiresText,
|
|
&record.Nonce, &record.TokenHash, &record.LeaseExpiresText, &record.ClaimedAt, &closed,
|
|
&record.CurrentAuthorizationTaskVersion, &record.CurrentGoodsID, &record.CurrentSKUColor,
|
|
&record.CurrentSKUSize, &record.CurrentQuantity, &record.CurrentTotalPriceCap,
|
|
&record.CurrentAuthorizationExpiresText,
|
|
&record.AuthorizationStatus, &record.CurrentAttemptGeneration, &record.AttemptStatus,
|
|
&record.TaskStatus, &record.CurrentTaskVersion,
|
|
&record.CurrentTaskTitle, &record.CurrentTaskGoodsID, &record.CurrentTaskSKUColor,
|
|
&record.CurrentTaskSKUSize, &record.CurrentTaskQuantity, &record.CurrentTaskMaxTotalPrice)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return claimRecord{}, false, nil
|
|
}
|
|
if err != nil {
|
|
return claimRecord{}, false, err
|
|
}
|
|
record.ClosedAt = closed.String
|
|
if !validUUID(record.AttemptID) || !validUUID(record.TaskID) || !validUUID(record.AuthorizationID) ||
|
|
!deviceauth.ValidDeviceID(record.DeviceID) || !validUUID(record.SessionID) || record.Generation <= 0 ||
|
|
record.CurrentAttemptGeneration != record.Generation ||
|
|
record.TaskVersion <= 0 || record.AuthorizationTaskVersion <= 0 || strings.TrimSpace(record.TaskTitle) == "" ||
|
|
!digitsOnly(record.GoodsID) || record.SKUColor == "" || record.SKUSize == "" || record.Quantity <= 0 ||
|
|
!canonicalMoney(record.TotalPriceCap) || len(record.Nonce) != sha256.Size || len(record.TokenHash) != sha256.Size {
|
|
return claimRecord{}, false, errors.New("stored task claim metadata is invalid")
|
|
}
|
|
record.LeaseExpiresAt, err = parseCanonicalTime(record.LeaseExpiresText)
|
|
if err != nil {
|
|
return claimRecord{}, false, errors.New("stored task claim lease is invalid")
|
|
}
|
|
record.AuthorizationExpiresAt, err = parseCanonicalTime(record.AuthorizationExpiresText)
|
|
if err != nil {
|
|
return claimRecord{}, false, errors.New("stored authorization expiry is invalid")
|
|
}
|
|
record.CurrentAuthorizationExpiresAt, err = parseCanonicalTime(record.CurrentAuthorizationExpiresText)
|
|
if err != nil {
|
|
return claimRecord{}, false, errors.New("current authorization expiry is invalid")
|
|
}
|
|
derived := deriveToken(store.secret, record.DeviceID, record.TaskID, record.AuthorizationID, record.AttemptID, record.Generation, record.Nonce)
|
|
if !matchingHash(tokenHash(derived), record.TokenHash) {
|
|
return claimRecord{}, false, errors.New("task claim secret does not match stored claim")
|
|
}
|
|
return record, true, nil
|
|
}
|
|
|
|
func (record claimRecord) authorizationConsistent() bool {
|
|
return record.AuthorizationTaskVersion == record.CurrentAuthorizationTaskVersion &&
|
|
record.GoodsID == record.CurrentGoodsID && record.SKUColor == record.CurrentSKUColor &&
|
|
record.SKUSize == record.CurrentSKUSize && record.Quantity == record.CurrentQuantity &&
|
|
record.TotalPriceCap == record.CurrentTotalPriceCap &&
|
|
record.AuthorizationExpiresText == record.CurrentAuthorizationExpiresText &&
|
|
record.TaskTitle == record.CurrentTaskTitle && record.GoodsID == record.CurrentTaskGoodsID &&
|
|
record.SKUColor == record.CurrentTaskSKUColor && record.SKUSize == record.CurrentTaskSKUSize &&
|
|
record.Quantity == record.CurrentTaskQuantity && record.TotalPriceCap == record.CurrentTaskMaxTotalPrice
|
|
}
|
|
|
|
func (record claimRecord) recoverableBusinessState() bool {
|
|
if record.TaskStatus == "CLAIMED" && record.AttemptStatus == "CLAIMED" {
|
|
return record.CurrentTaskVersion == record.TaskVersion
|
|
}
|
|
// A later server task may advance this same attempt to ORDERING. A valid lease and identical
|
|
// ownership recover that attempt; claim-next still cannot select another task.
|
|
return record.TaskStatus == "ORDERING" && record.AttemptStatus == "ORDERING" &&
|
|
record.TaskVersion < math.MaxInt && record.CurrentTaskVersion == record.TaskVersion+1
|
|
}
|
|
|
|
func (store *Store) responseFor(record claimRecord, responseLease string) (ClaimResponse, error) {
|
|
if !validCanonicalTime(responseLease) {
|
|
return ClaimResponse{}, errors.New("stored claim response lease is invalid")
|
|
}
|
|
token := deriveToken(store.secret, record.DeviceID, record.TaskID, record.AuthorizationID, record.AttemptID, record.Generation, record.Nonce)
|
|
return ClaimResponse{
|
|
Task: ClaimedTask{ID: record.TaskID, Version: record.TaskVersion, Title: record.TaskTitle,
|
|
ProductURL: productURL(record.GoodsID), GoodsID: record.GoodsID, SKUColor: record.SKUColor,
|
|
SKUSize: record.SKUSize, Quantity: record.Quantity, MaxTotalPrice: record.TotalPriceCap},
|
|
Authorization: ClaimedAuthorization{ID: record.AuthorizationID, TaskVersion: record.AuthorizationTaskVersion, ExpiresAt: record.AuthorizationExpiresText},
|
|
Attempt: ClaimedAttempt{ID: record.AttemptID, ClaimToken: hex.EncodeToString(token), ClaimGeneration: record.Generation, LeaseExpiresAt: responseLease},
|
|
}, nil
|
|
}
|
|
|
|
type candidate struct {
|
|
AuthorizationID, TaskID, Title, GoodsID, SKUColor, SKUSize, TotalPriceCap string
|
|
TaskVersion, Quantity int
|
|
AuthorizationExpiresText string
|
|
AuthorizationExpiresAt time.Time
|
|
}
|
|
|
|
func findCandidate(ctx context.Context, transaction *sql.Tx, now time.Time) (candidate, bool, error) {
|
|
rows, err := transaction.QueryContext(ctx, `SELECT authorizations.id, tasks.id, tasks.version,
|
|
tasks.title, tasks.goods_id, tasks.sku_color, tasks.sku_size, tasks.quantity,
|
|
tasks.max_total_price, authorizations.expires_at
|
|
FROM order_authorizations AS authorizations
|
|
JOIN tasks ON tasks.id = authorizations.task_id
|
|
WHERE authorizations.status = 'ACTIVE' AND tasks.status = 'PENDING'
|
|
AND authorizations.task_version = tasks.version
|
|
AND authorizations.goods_id = tasks.goods_id
|
|
AND authorizations.sku_color = tasks.sku_color
|
|
AND authorizations.sku_size = tasks.sku_size
|
|
AND authorizations.quantity = tasks.quantity
|
|
AND authorizations.total_price_cap = tasks.max_total_price
|
|
ORDER BY authorizations.created_at, authorizations.rowid, authorizations.id`)
|
|
if err != nil {
|
|
return candidate{}, false, err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var item candidate
|
|
if err := rows.Scan(&item.AuthorizationID, &item.TaskID, &item.TaskVersion, &item.Title,
|
|
&item.GoodsID, &item.SKUColor, &item.SKUSize, &item.Quantity, &item.TotalPriceCap,
|
|
&item.AuthorizationExpiresText); err != nil {
|
|
return candidate{}, false, err
|
|
}
|
|
item.AuthorizationExpiresAt, err = parseCanonicalTime(item.AuthorizationExpiresText)
|
|
if err != nil {
|
|
return candidate{}, false, errors.New("stored authorization expiry is invalid")
|
|
}
|
|
if !validCandidate(item) {
|
|
return candidate{}, false, errors.New("stored claim candidate is invalid")
|
|
}
|
|
if item.AuthorizationExpiresAt.After(now) {
|
|
if err := rows.Close(); err != nil {
|
|
return candidate{}, false, err
|
|
}
|
|
return item, true, nil
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return candidate{}, false, err
|
|
}
|
|
return candidate{}, false, nil
|
|
}
|
|
|
|
func validCandidate(item candidate) bool {
|
|
return validUUID(item.AuthorizationID) && validUUID(item.TaskID) && item.TaskVersion > 0 && item.TaskVersion < math.MaxInt &&
|
|
strings.TrimSpace(item.Title) != "" && digitsOnly(item.GoodsID) && item.SKUColor != "" && item.SKUSize != "" &&
|
|
item.Quantity > 0 && canonicalMoney(item.TotalPriceCap)
|
|
}
|
|
|
|
func validAttemptStatus(value sql.NullString) bool {
|
|
return value.Valid && oneOf(value.String, "CLAIMED", "ORDERING", "FAILED", "FENCED", "ABANDONED")
|
|
}
|
|
|
|
func validAuthorizationStatus(value sql.NullString) bool {
|
|
return value.Valid && oneOf(value.String, "ACTIVE", "CLAIMED", "FENCED", "CONSUMED", "EXPIRED", "ABANDONED")
|
|
}
|
|
|
|
func validTaskStatus(value sql.NullString) bool {
|
|
return value.Valid && oneOf(value.String, "DRAFT", "PENDING", "CLAIMED", "ORDERING", "NEEDS_MANUAL",
|
|
"WAITING_PAYMENT", "RECONCILIATION_REQUIRED", "SUCCEEDED", "FAILED", "CANCELED")
|
|
}
|
|
|
|
func oneOf(value string, allowed ...string) bool {
|
|
for _, item := range allowed {
|
|
if value == item {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func nextGeneration(ctx context.Context, transaction *sql.Tx, taskID string) (int, error) {
|
|
var maximum int64
|
|
if err := transaction.QueryRowContext(ctx, `SELECT COALESCE(MAX(claim_generation), 0) FROM purchase_attempts WHERE task_id = ?`, taskID).Scan(&maximum); err != nil {
|
|
return 0, err
|
|
}
|
|
if maximum < 0 || maximum >= int64(math.MaxInt) {
|
|
return 0, errors.New("task claim generation is exhausted")
|
|
}
|
|
return int(maximum) + 1, nil
|
|
}
|
|
|
|
func (store *Store) serverNow() (time.Time, error) {
|
|
now := store.now().UTC()
|
|
if now.IsZero() {
|
|
return time.Time{}, errors.New("task claim clock is invalid")
|
|
}
|
|
return now, nil
|
|
}
|
|
|
|
func (store *Store) randomBytes(size int) ([]byte, error) {
|
|
value := make([]byte, size)
|
|
store.randomMu.Lock()
|
|
_, err := io.ReadFull(store.random, value)
|
|
store.randomMu.Unlock()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("generate task claim randomness: %w", err)
|
|
}
|
|
return value, nil
|
|
}
|
|
|
|
func (store *Store) newUUID() (string, error) {
|
|
value, err := store.randomBytes(16)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
value[6] = (value[6] & 0x0f) | 0x40
|
|
value[8] = (value[8] & 0x3f) | 0x80
|
|
encoded := hex.EncodeToString(value)
|
|
return encoded[:8] + "-" + encoded[8:12] + "-" + encoded[12:16] + "-" + encoded[16:20] + "-" + encoded[20:], nil
|
|
}
|
|
|
|
func exactlyOne(result sql.Result) (bool, error) {
|
|
rows, err := result.RowsAffected()
|
|
return rows == 1, err
|
|
}
|
|
|
|
func formatTime(value time.Time) string { return value.UTC().Format(time.RFC3339Nano) }
|
|
|
|
func parseCanonicalTime(value string) (time.Time, error) {
|
|
if !strings.HasSuffix(value, "Z") || strings.TrimSpace(value) != value {
|
|
return time.Time{}, ErrInvalid
|
|
}
|
|
parsed, err := time.Parse(time.RFC3339Nano, value)
|
|
if err != nil || parsed.Location() != time.UTC || formatTime(parsed) != value {
|
|
return time.Time{}, ErrInvalid
|
|
}
|
|
return parsed, nil
|
|
}
|
|
|
|
func validCanonicalTime(value string) bool {
|
|
_, err := parseCanonicalTime(value)
|
|
return err == nil
|
|
}
|
|
|
|
func validUUID(value string) bool {
|
|
if len(value) != 36 {
|
|
return false
|
|
}
|
|
for index, character := range value {
|
|
if index == 8 || index == 13 || index == 18 || index == 23 {
|
|
if character != '-' {
|
|
return false
|
|
}
|
|
continue
|
|
}
|
|
if !(character >= '0' && character <= '9' || character >= 'a' && character <= 'f') {
|
|
return false
|
|
}
|
|
}
|
|
return value[14] == '4' && (value[19] == '8' || value[19] == '9' || value[19] == 'a' || value[19] == 'b')
|
|
}
|
|
|
|
func digitsOnly(value string) bool {
|
|
if value == "" {
|
|
return false
|
|
}
|
|
for _, character := range value {
|
|
if character < '0' || character > '9' {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func canonicalMoney(value string) bool {
|
|
parts := strings.Split(value, ".")
|
|
if len(parts) != 2 || len(parts[0]) == 0 || len(parts[1]) != 2 || (len(parts[0]) > 1 && parts[0][0] == '0') {
|
|
return false
|
|
}
|
|
for _, part := range parts {
|
|
if !digitsOnly(part) {
|
|
return false
|
|
}
|
|
}
|
|
cents := new(big.Int)
|
|
_, ok := cents.SetString(parts[0]+parts[1], 10)
|
|
return ok && cents.Sign() > 0
|
|
}
|
|
|
|
func productURL(goodsID string) string {
|
|
return "https://mobile.yangkeduo.com/goods.html?goods_id=" + goodsID
|
|
}
|