package sqlite import ( "context" "database/sql" "errors" "time" "cmroubao/backend-api/internal/domain" "cmroubao/backend-api/internal/usecase" ) const createProcurementTaskOperation = "CREATE_PROCUREMENT_TASK" func (store *Store) GetProcurementSourceItem( ctx context.Context, creatorSubject, itemID string, ) (domain.ProcurementSourceItem, error) { item, err := scanFreightOrderItem(store.db.QueryRowContext( ctx, `SELECT item.id, item.freight_order_id, item.external_item_id, item.title, item.product_spec, item.sku, item.quantity, item.product_thumb_ref, item.original_unit_price_minor, item.original_currency, item.purchase_status, item.canonical_sha256, item.revision, item.is_present, item.first_sync_run_id, item.last_sync_run_id, item.created_at, item.updated_at FROM freight_order_items AS item JOIN freight_orders AS freight ON freight.id = item.freight_order_id WHERE freight.creator_subject = ? AND item.id = ? AND item.is_present = 1`, creatorSubject, itemID, )) if errors.Is(err, sql.ErrNoRows) { return domain.ProcurementSourceItem{}, usecase.ErrRepositoryNotFound } if err != nil { return domain.ProcurementSourceItem{}, repositoryFailure(err) } var canceled sql.NullBool if err := store.db.QueryRowContext( ctx, `SELECT freight.is_canceled FROM freight_orders AS freight JOIN freight_order_items AS item ON item.freight_order_id = freight.id WHERE freight.creator_subject = ? AND item.id = ?`, creatorSubject, itemID, ).Scan(&canceled); err != nil { return domain.ProcurementSourceItem{}, repositoryFailure(err) } var isCanceled *bool if canceled.Valid { isCanceled = &canceled.Bool } return domain.ProcurementSourceItem{ Item: item, IsCanceled: isCanceled, }, nil } func (store *Store) CreateProcurementRequest( ctx context.Context, candidate domain.ProcurementRequest, ) (domain.ProcurementRequest, bool, error) { tx, err := store.db.BeginTx(ctx, nil) if err != nil { return domain.ProcurementRequest{}, false, repositoryFailure(err) } defer tx.Rollback() existing, err := getProcurementRequestBySource( ctx, tx, candidate.CreatorSubject, candidate.FreightOrderItemID, candidate.SourceRevision, ) if err == nil { if err := tx.Commit(); err != nil { return domain.ProcurementRequest{}, false, repositoryFailure(err) } return existing, false, nil } if !errors.Is(err, usecase.ErrRepositoryNotFound) { return domain.ProcurementRequest{}, false, err } var currentRevision int var currentSHA string var present bool if err := tx.QueryRowContext( ctx, `SELECT item.revision, item.canonical_sha256, item.is_present FROM freight_order_items AS item JOIN freight_orders AS freight ON freight.id = item.freight_order_id WHERE freight.creator_subject = ? AND item.id = ?`, candidate.CreatorSubject, candidate.FreightOrderItemID, ).Scan(¤tRevision, ¤tSHA, &present); err != nil { if errors.Is(err, sql.ErrNoRows) { return domain.ProcurementRequest{}, false, usecase.ErrRepositoryNotFound } return domain.ProcurementRequest{}, false, repositoryFailure(err) } if !present || currentRevision != candidate.SourceRevision || currentSHA != candidate.SourceSHA256 { return domain.ProcurementRequest{}, false, usecase.ErrProcurementSourceChanged } _, err = tx.ExecContext( ctx, `INSERT INTO procurement_requests ( id, creator_subject, freight_order_item_id, source_revision, source_sha256, title, product_spec, sku, quantity, source_purchase_status, source_is_canceled, procurement_confirmed_by_user_id, procurement_confirmed_at, status, blocking_code, reference_asset_id, purchase_task_id, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?, ?)`, candidate.ID, candidate.CreatorSubject, candidate.FreightOrderItemID, candidate.SourceRevision, candidate.SourceSHA256, candidate.Title, candidate.ProductSpec, candidate.SKU, nullableFreightQuantity(candidate.Quantity), nullableString(candidate.SourcePurchaseStatus), nullableBool(candidate.SourceIsCanceled), candidate.ProcurementConfirmedByUserID, formatTimestamp(candidate.ProcurementConfirmedAt), candidate.Status, nullableString(candidate.BlockingCode), formatTimestamp(candidate.CreatedAt), formatTimestamp(candidate.UpdatedAt), ) if err != nil { return domain.ProcurementRequest{}, false, repositoryFailure(err) } if err := tx.Commit(); err != nil { return domain.ProcurementRequest{}, false, repositoryFailure(err) } return candidate, true, nil } func (store *Store) ListProcurementRequestsForOrder( ctx context.Context, creatorSubject, orderID string, ) ([]domain.ProcurementRequest, error) { rows, err := store.db.QueryContext( ctx, procurementRequestSelect+` WHERE request.creator_subject = ? AND item.freight_order_id = ? ORDER BY item.id, request.source_revision DESC, request.id DESC`, creatorSubject, orderID, ) if err != nil { return nil, repositoryFailure(err) } defer rows.Close() requests := make([]domain.ProcurementRequest, 0) for rows.Next() { request, err := scanProcurementRequest(rows) if err != nil { return nil, repositoryFailure(err) } requests = append(requests, request) } if err := rows.Err(); err != nil { return nil, repositoryFailure(err) } return requests, nil } func (store *Store) GetProcurementRequest( ctx context.Context, creatorSubject, requestID string, ) (domain.ProcurementRequest, error) { return getProcurementRequest(ctx, store.db, creatorSubject, requestID) } func (store *Store) BindProcurementReference( ctx context.Context, creatorSubject, requestID, assetID string, now time.Time, ) (domain.ProcurementRequest, error) { tx, err := store.db.BeginTx(ctx, nil) if err != nil { return domain.ProcurementRequest{}, repositoryFailure(err) } defer tx.Rollback() request, err := getProcurementRequest( ctx, tx, creatorSubject, requestID, ) if err != nil { return domain.ProcurementRequest{}, err } if request.SourceChanged { if request.PurchaseTaskID == nil { if err := markProcurementSourceChanged(ctx, tx, request.ID, now); err != nil { return domain.ProcurementRequest{}, err } } if err := tx.Commit(); err != nil { return domain.ProcurementRequest{}, repositoryFailure(err) } return domain.ProcurementRequest{}, usecase.ErrProcurementSourceChanged } if request.ReferenceAssetID != nil && *request.ReferenceAssetID == assetID && (request.Status == domain.ProcurementReady || request.Status == domain.ProcurementTaskCreated) { if err := tx.Commit(); err != nil { return domain.ProcurementRequest{}, repositoryFailure(err) } return request, nil } if request.Status != domain.ProcurementNeedsImage { return domain.ProcurementRequest{}, usecase.ErrProcurementStateConflict } var available int if err := tx.QueryRowContext( ctx, `SELECT EXISTS ( SELECT 1 FROM assets AS asset WHERE asset.id = ? AND asset.creator_subject = ? AND asset.purpose = 'TASK_REFERENCE' AND NOT EXISTS ( SELECT 1 FROM purchase_tasks AS task WHERE task.image_asset_id = asset.id ) AND NOT EXISTS ( SELECT 1 FROM procurement_requests AS other WHERE other.reference_asset_id = asset.id AND other.id != ? ) )`, assetID, creatorSubject, requestID, ).Scan(&available); err != nil { return domain.ProcurementRequest{}, repositoryFailure(err) } if available != 1 { return domain.ProcurementRequest{}, usecase.ErrAssetUnavailable } if _, err := tx.ExecContext( ctx, `UPDATE procurement_requests SET reference_asset_id = ?, status = 'READY', updated_at = ? WHERE id = ? AND status = 'NEEDS_IMAGE'`, assetID, formatTimestamp(now), requestID, ); err != nil { return domain.ProcurementRequest{}, repositoryFailure(err) } updated, err := getProcurementRequest(ctx, tx, creatorSubject, requestID) if err != nil { return domain.ProcurementRequest{}, err } if err := tx.Commit(); err != nil { return domain.ProcurementRequest{}, repositoryFailure(err) } return updated, nil } func (store *Store) CreateProcurementTask( ctx context.Context, expected domain.ProcurementRequest, task domain.PurchaseTask, event domain.TaskEvent, source domain.PurchaseTaskSource, idempotencyKey, requestSHA256 string, ) (domain.PurchaseTask, bool, error) { tx, err := store.db.BeginTx(ctx, nil) if err != nil { return domain.PurchaseTask{}, false, repositoryFailure(err) } defer tx.Rollback() existingHash, resourceID, found, err := lookupIdempotency( ctx, tx, task.CreatorSubject, createProcurementTaskOperation, idempotencyKey, ) if err != nil { return domain.PurchaseTask{}, false, err } if found { if existingHash != requestSHA256 { return domain.PurchaseTask{}, false, usecase.ErrIdempotencyConflict } existing, err := getTaskByID( ctx, tx, task.CreatorSubject, resourceID, ) if err != nil { return domain.PurchaseTask{}, false, err } if err := tx.Commit(); err != nil { return domain.PurchaseTask{}, false, repositoryFailure(err) } return existing, false, nil } request, err := getProcurementRequest( ctx, tx, task.CreatorSubject, expected.ID, ) if err != nil { return domain.PurchaseTask{}, false, err } if request.Status == domain.ProcurementTaskCreated && request.PurchaseTaskID != nil { existing, err := getTaskByID( ctx, tx, task.CreatorSubject, *request.PurchaseTaskID, ) if err != nil { return domain.PurchaseTask{}, false, err } if err := insertIdempotency( ctx, tx, task.CreatorSubject, createProcurementTaskOperation, idempotencyKey, requestSHA256, "PURCHASE_TASK", existing.ID, task.CreatedAt, ); err != nil { return domain.PurchaseTask{}, false, err } if err := tx.Commit(); err != nil { return domain.PurchaseTask{}, false, repositoryFailure(err) } return existing, false, nil } if request.SourceChanged { if request.PurchaseTaskID == nil { if err := markProcurementSourceChanged( ctx, tx, request.ID, task.CreatedAt, ); err != nil { return domain.PurchaseTask{}, false, err } } if err := tx.Commit(); err != nil { return domain.PurchaseTask{}, false, repositoryFailure(err) } return domain.PurchaseTask{}, false, usecase.ErrProcurementSourceChanged } if request.Status != domain.ProcurementReady || request.ReferenceAssetID == nil || request.SourceSHA256 != expected.SourceSHA256 { return domain.PurchaseTask{}, false, usecase.ErrProcurementStateConflict } var assetAvailable int if err := tx.QueryRowContext( ctx, `SELECT EXISTS ( SELECT 1 FROM assets AS asset WHERE asset.id = ? AND asset.creator_subject = ? AND asset.purpose = 'TASK_REFERENCE' AND NOT EXISTS ( SELECT 1 FROM purchase_tasks AS other WHERE other.image_asset_id = asset.id ) )`, *request.ReferenceAssetID, task.CreatorSubject, ).Scan(&assetAvailable); err != nil { return domain.PurchaseTask{}, false, repositoryFailure(err) } if assetAvailable != 1 { return domain.PurchaseTask{}, false, usecase.ErrAssetUnavailable } _, err = tx.ExecContext( ctx, `INSERT INTO purchase_tasks ( id, creator_subject, created_by_user_id, source_ref, title, description, sku, image_asset_id, quantity, max_budget_cents, currency, status, version, cancel_reason, canceled_at, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, 'CNY', 'PENDING', 1, NULL, NULL, ?, ?)`, task.ID, task.CreatorSubject, nullableString(task.CreatedByUserID), nullableString(task.SourceRef), task.Title, task.Description, task.SKU, task.ImageAssetID, task.Quantity, formatTimestamp(task.CreatedAt), formatTimestamp(task.UpdatedAt), ) if err != nil { return domain.PurchaseTask{}, false, repositoryFailure(err) } if err := insertTaskEvent(ctx, tx, event); err != nil { return domain.PurchaseTask{}, false, err } _, err = tx.ExecContext( ctx, `INSERT INTO purchase_task_sources ( task_id, procurement_request_id, freight_order_item_id, source_revision, source_sha256, created_at ) VALUES (?, ?, ?, ?, ?, ?)`, source.TaskID, source.ProcurementRequestID, source.FreightOrderItemID, source.SourceRevision, source.SourceSHA256, formatTimestamp(source.CreatedAt), ) if err != nil { return domain.PurchaseTask{}, false, repositoryFailure(err) } result, err := tx.ExecContext( ctx, `UPDATE procurement_requests SET status = 'TASK_CREATED', purchase_task_id = ?, updated_at = ? WHERE id = ? AND status = 'READY'`, task.ID, formatTimestamp(task.CreatedAt), request.ID, ) if err != nil { return domain.PurchaseTask{}, false, repositoryFailure(err) } changed, err := result.RowsAffected() if err != nil { return domain.PurchaseTask{}, false, repositoryFailure(err) } if changed != 1 { return domain.PurchaseTask{}, false, usecase.ErrProcurementStateConflict } if err := insertIdempotency( ctx, tx, task.CreatorSubject, createProcurementTaskOperation, idempotencyKey, requestSHA256, "PURCHASE_TASK", task.ID, task.CreatedAt, ); err != nil { return domain.PurchaseTask{}, false, err } if err := tx.Commit(); err != nil { return domain.PurchaseTask{}, false, repositoryFailure(err) } return task, true, nil } const procurementRequestSelect = `SELECT request.id, request.creator_subject, request.freight_order_item_id, request.source_revision, request.source_sha256, request.title, request.product_spec, request.sku, request.quantity, request.source_purchase_status, request.source_is_canceled, request.procurement_confirmed_by_user_id, request.procurement_confirmed_at, request.status, request.blocking_code, request.reference_asset_id, request.purchase_task_id, request.created_at, request.updated_at, CASE WHEN item.revision != request.source_revision OR item.canonical_sha256 != request.source_sha256 OR item.is_present != 1 OR COALESCE(freight.is_canceled, -1) != COALESCE(request.source_is_canceled, -1) THEN 1 ELSE 0 END FROM procurement_requests AS request JOIN freight_order_items AS item ON item.id = request.freight_order_item_id JOIN freight_orders AS freight ON freight.id = item.freight_order_id ` func getProcurementRequest( ctx context.Context, query queryRower, creatorSubject, requestID string, ) (domain.ProcurementRequest, error) { request, err := scanProcurementRequest(query.QueryRowContext( ctx, procurementRequestSelect+` WHERE request.creator_subject = ? AND request.id = ?`, creatorSubject, requestID, )) if errors.Is(err, sql.ErrNoRows) { return domain.ProcurementRequest{}, usecase.ErrRepositoryNotFound } if err != nil { return domain.ProcurementRequest{}, repositoryFailure(err) } return request, nil } func getProcurementRequestBySource( ctx context.Context, query queryRower, creatorSubject, itemID string, revision int, ) (domain.ProcurementRequest, error) { request, err := scanProcurementRequest(query.QueryRowContext( ctx, procurementRequestSelect+` WHERE request.creator_subject = ? AND request.freight_order_item_id = ? AND request.source_revision = ?`, creatorSubject, itemID, revision, )) if errors.Is(err, sql.ErrNoRows) { return domain.ProcurementRequest{}, usecase.ErrRepositoryNotFound } if err != nil { return domain.ProcurementRequest{}, repositoryFailure(err) } return request, nil } func scanProcurementRequest( scanner rowScanner, ) (domain.ProcurementRequest, error) { var request domain.ProcurementRequest var quantity sql.NullInt64 var purchaseStatus sql.NullString var isCanceled sql.NullBool var blockingCode, assetID, taskID sql.NullString var confirmedAt, createdAt, updatedAt string var sourceChanged bool err := scanner.Scan( &request.ID, &request.CreatorSubject, &request.FreightOrderItemID, &request.SourceRevision, &request.SourceSHA256, &request.Title, &request.ProductSpec, &request.SKU, &quantity, &purchaseStatus, &isCanceled, &request.ProcurementConfirmedByUserID, &confirmedAt, &request.Status, &blockingCode, &assetID, &taskID, &createdAt, &updatedAt, &sourceChanged, ) if err != nil { return domain.ProcurementRequest{}, err } if quantity.Valid { value := int(quantity.Int64) request.Quantity = &value } request.SourcePurchaseStatus = optionalString(purchaseStatus) if isCanceled.Valid { request.SourceIsCanceled = &isCanceled.Bool } request.BlockingCode = optionalString(blockingCode) request.ReferenceAssetID = optionalString(assetID) request.PurchaseTaskID = optionalString(taskID) request.SourceChanged = sourceChanged request.ProcurementConfirmedAt, err = parseTimestamp(confirmedAt) if err != nil { return domain.ProcurementRequest{}, err } request.CreatedAt, err = parseTimestamp(createdAt) if err != nil { return domain.ProcurementRequest{}, err } request.UpdatedAt, err = parseTimestamp(updatedAt) return request, err } func markProcurementSourceChanged( ctx context.Context, tx *sql.Tx, requestID string, now time.Time, ) error { _, err := tx.ExecContext( ctx, `UPDATE procurement_requests SET status = 'SOURCE_CHANGED', blocking_code = 'SOURCE_CHANGED', updated_at = ? WHERE id = ? AND purchase_task_id IS NULL`, formatTimestamp(now), requestID, ) if err != nil { return repositoryFailure(err) } return nil } var _ usecase.ProcurementRepository = (*Store)(nil)