chengma/cmroubao
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(t206): connect Android procurement tasks

Browse Source
This commit is contained in:
QiuSW
2026-07-27 11:11:16 +08:00
parent 9e698c1b6c
commit 45b66eeea7
35 changed files with 2634 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-api/internal/config/config.go
+3 -3
View File
@@ -23,7 +23,7 @@ const (
defaultDatabasePath = "var/cmroubao.db"
defaultAssetDirectory = "var/assets"
defaultClaimLease = 10 * time.Minute
defaultRunningLease = 90 * time.Second
defaultRunningLease = 30 * time.Minute
defaultReadinessTTL = 2 * time.Minute
)
@@ -121,8 +121,8 @@ func Load(lookup LookupEnvironment) (Config, error) {
lookup,
RunningLeaseEnvironment,
defaultRunningLease,
30*time.Second,
10*time.Minute,
5*time.Minute,
120*time.Minute,
)
if err != nil {
return Config{}, err
backend-api/internal/config/config_test.go
+10 -4
View File
@@ -30,7 +30,7 @@ func TestLoadUsesSafeDefaults(t *testing.T) {
t.Fatal("server safety limits must all be positive")
}
if cfg.ClaimLease != 10*time.Minute ||
cfg.RunningLease != 90*time.Second ||
cfg.RunningLease != 30*time.Minute ||
cfg.ReadinessTTL != 2*time.Minute {
t.Fatalf(
"lifecycle durations = %s / %s / %s",
@@ -52,7 +52,7 @@ func TestLoadAcceptsExplicitConfiguration(t *testing.T) {
TLSCertificateEnvironment: "tmp/server.crt",
TLSPrivateKeyEnvironment: "tmp/server.key",
ClaimLeaseEnvironment: "15m",
RunningLeaseEnvironment: "2m",
RunningLeaseEnvironment: "45m",
ReadinessTTLEnvironment: "3m",
}
@@ -79,7 +79,7 @@ func TestLoadAcceptsExplicitConfiguration(t *testing.T) {
)
}
if cfg.ClaimLease != 15*time.Minute ||
cfg.RunningLease != 2*time.Minute ||
cfg.RunningLease != 45*time.Minute ||
cfg.ReadinessTTL != 3*time.Minute {
t.Fatalf(
"lifecycle durations = %s / %s / %s",
@@ -174,10 +174,16 @@ func TestLoadRejectsUnsafeOrInvalidValues(t *testing.T) {
ClaimLeaseEnvironment: "59s",
},
},
{
name: "running lease below minimum",
values: map[string]string{
RunningLeaseEnvironment: "4m59s",
},
},
{
name: "running lease above maximum",
values: map[string]string{
RunningLeaseEnvironment: "11m",
RunningLeaseEnvironment: "121m",
},
},
{
backend-api/internal/repository/sqlite/lifecycle_repository.go
+35 -13
View File
@@ -498,13 +498,12 @@ func (s *Store) HeartbeatTask(
if err != nil {
return usecase.TaskHeartbeatRepositoryResult{}, err
}
if err := validateClaim(
if err := validateClaimOwner(
task,
request.UserID,
request.DeviceID,
request.ClaimGeneration,
request.ClaimTokenHash,
request.Now,
); err != nil {
return usecase.TaskHeartbeatRepositoryResult{}, err
}
@@ -512,6 +511,11 @@ func (s *Store) HeartbeatTask(
return usecase.TaskHeartbeatRepositoryResult{},
usecase.ErrTaskStateConflict
}
expired := !task.ClaimExpiresAt.After(request.Now)
if expired && request.Step != "SAFE_STOPPED" {
return usecase.TaskHeartbeatRepositoryResult{},
usecase.ErrClaimExpired
}
execution, err := getExecutionByID(ctx, tx, request.ExecutionID)
if err != nil {
return usecase.TaskHeartbeatRepositoryResult{}, err
@@ -525,7 +529,8 @@ func (s *Store) HeartbeatTask(
usecase.ErrExecutionMismatch
}
expiresAt := *task.ClaimExpiresAt
if request.MinimumExpiry.After(expiresAt) {
if !expired &&
request.MinimumExpiry.After(expiresAt) {
expiresAt = request.MinimumExpiry
}
result, err := tx.ExecContext(
@@ -561,8 +566,7 @@ func (s *Store) HeartbeatTask(
AND claimed_by_user_id = ?
AND claimed_by_device_id = ?
AND claim_generation = ?
AND claim_token_hash = ?
AND claim_expires_at > ?`,
AND claim_token_hash = ?`,
formatTimestamp(expiresAt),
formatTimestamp(request.Now),
request.TaskID,
@@ -570,7 +574,6 @@ func (s *Store) HeartbeatTask(
request.DeviceID,
request.ClaimGeneration,
request.ClaimTokenHash,
formatTimestamp(request.Now),
)
if err != nil {
return usecase.TaskHeartbeatRepositoryResult{}, repositoryFailure(err)
@@ -581,7 +584,7 @@ func (s *Store) HeartbeatTask(
}
if affected != 1 {
return usecase.TaskHeartbeatRepositoryResult{},
usecase.ErrClaimExpired
usecase.ErrTaskVersionConflict
}
_, err = tx.ExecContext(
ctx,
@@ -797,13 +800,12 @@ func (s *Store) AcknowledgeTaskCancellation(
if err != nil {
return domain.PurchaseTask{}, false, err
}
if err := validateClaim(
if err := validateClaimOwner(
task,
request.UserID,
request.DeviceID,
request.ClaimGeneration,
request.ClaimTokenHash,
request.Now,
); err != nil {
return domain.PurchaseTask{}, false, err
}
@@ -846,8 +848,7 @@ func (s *Store) AcknowledgeTaskCancellation(
AND claimed_by_user_id = ?
AND claimed_by_device_id = ?
AND claim_generation = ?
AND claim_token_hash = ?
AND claim_expires_at > ?`,
AND claim_token_hash = ?`,
formatTimestamp(request.Now),
formatTimestamp(request.Now),
request.TaskID,
@@ -856,7 +857,6 @@ func (s *Store) AcknowledgeTaskCancellation(
request.DeviceID,
request.ClaimGeneration,
request.ClaimTokenHash,
formatTimestamp(request.Now),
)
if err != nil {
return domain.PurchaseTask{}, false, repositoryFailure(err)
@@ -1041,6 +1041,28 @@ func validateClaim(
generation int64,
tokenHash string,
now time.Time,
) error {
if err := validateClaimOwner(
task,
userID,
deviceID,
generation,
tokenHash,
); err != nil {
return err
}
if task.ClaimExpiresAt == nil || !task.ClaimExpiresAt.After(now) {
return usecase.ErrClaimExpired
}
return nil
}
func validateClaimOwner(
task domain.PurchaseTask,
userID string,
deviceID string,
generation int64,
tokenHash string,
) error {
if task.ClaimedByUserID == nil ||
*task.ClaimedByUserID != userID ||
@@ -1051,7 +1073,7 @@ func validateClaim(
*task.ClaimTokenHash != tokenHash {
return usecase.ErrClaimInvalid
}
if task.ClaimExpiresAt == nil || !task.ClaimExpiresAt.After(now) {
if task.ClaimExpiresAt == nil {
return usecase.ErrClaimExpired
}
return nil
backend-api/internal/repository/sqlite/lifecycle_repository_test.go
+28 -4
View File
@@ -748,6 +748,29 @@ func TestLifecycleRepositoryHeartbeatExtendsLeaseWithoutEvent(
},
)
assertLifecycleError(t, err, usecase.ErrClaimExpired)
expiredHeartbeat, err := fixture.store.HeartbeatTask(
ctx,
usecase.TaskHeartbeatRepositoryRequest{
UserID: fixture.buyerOneID,
DeviceID: fixture.deviceOneID,
TaskID: claimed.ID,
ExecutionID: started.Execution.ID,
ClaimGeneration: claimed.ClaimGeneration,
ClaimTokenHash: tokenHash,
Step: "SAFE_STOPPED",
Now: expiredAt,
MinimumExpiry: expiredAt.Add(90 * time.Second),
},
)
if err != nil {
t.Fatalf("expired HeartbeatTask() error = %v", err)
}
if expiredHeartbeat.Task.ClaimExpiresAt == nil ||
!expiredHeartbeat.Task.ClaimExpiresAt.Equal(expiredAt) ||
expiredHeartbeat.Execution.CurrentStep != "SAFE_STOPPED" {
t.Fatalf("expired heartbeat = %+v", expiredHeartbeat)
}
}
func TestLifecycleRepositoryReleaseIsIdempotentAndInvalidatesClaim(
@@ -870,6 +893,7 @@ func TestLifecycleRepositoryCancelAcknowledgementEndsExecution(
if err != nil {
t.Fatalf("StartTask() error = %v", err)
}
expiredAt := *started.Task.ClaimExpiresAt
cancelAt := startRequest.Now.Add(10 * time.Second)
adminID := fixture.adminUserID
cancelRequested, err := fixture.store.CancelTask(
@@ -940,16 +964,16 @@ func TestLifecycleRepositoryCancelAcknowledgementEndsExecution(
ExecutionID: started.Execution.ID,
ClaimGeneration: claimed.ClaimGeneration,
ClaimTokenHash: tokenHash,
Step: "STOPPING",
Now: cancelAt.Add(time.Second),
MinimumExpiry: cancelAt.Add(91 * time.Second),
Step: "SAFE_STOPPED",
Now: expiredAt,
MinimumExpiry: expiredAt.Add(90 * time.Second),
},
)
if err != nil || !heartbeat.CancelRequested {
t.Fatalf("cancel heartbeat = %+v, error = %v", heartbeat, err)
}
ackAt := cancelAt.Add(2 * time.Second)
ackAt := expiredAt.Add(time.Second)
request := usecase.CancelAcknowledgementRepositoryRequest{
UserID: fixture.buyerOneID,
DeviceID: fixture.deviceOneID,
backend-api/internal/transport/httpapi/device_handlers.go
+20 -4
View File
@@ -5,6 +5,7 @@ import (
"net/http"
"strconv"
"strings"
"time"
"cmroubao/backend-api/internal/domain"
"cmroubao/backend-api/internal/transport/authcommon"
@@ -235,8 +236,11 @@ func (handler *deviceHandlers) startTask(ctx *gin.Context) {
}
ctx.Header("Cache-Control", "no-store")
ctx.JSON(http.StatusOK, gin.H{
"task": deviceTaskResponse(result.Task),
"execution": executionResponse(result.Execution),
"task": deviceTaskResponse(result.Task),
"execution": deviceExecutionResponse(
result.Execution,
result.Task.ClaimExpiresAt,
),
"replayed": result.Replayed,
"server_time": formatTime(result.ServerTime),
})
@@ -275,8 +279,11 @@ func (handler *deviceHandlers) heartbeatTask(ctx *gin.Context) {
}
ctx.Header("Cache-Control", "no-store")
ctx.JSON(http.StatusOK, gin.H{
"task": deviceTaskResponse(result.Task),
"execution": executionResponse(result.Execution),
"task": deviceTaskResponse(result.Task),
"execution": deviceExecutionResponse(
result.Execution,
result.Task.ClaimExpiresAt,
),
"cancel_requested": result.CancelRequested,
"server_time": formatTime(result.ServerTime),
})
@@ -453,3 +460,12 @@ func deviceTaskResponse(task domain.PurchaseTask) gin.H {
"currency": task.Currency,
}
}
func deviceExecutionResponse(
execution domain.TaskExecution,
expiresAt *time.Time,
) gin.H {
response := executionResponse(execution)
response["execution_expires_at"] = formatOptionalTime(expiresAt)
return response
}
backend-api/internal/transport/httpapi/device_handlers_test.go
+11 -3
View File
@@ -432,9 +432,14 @@ func TestDeviceStartAndTaskHeartbeatUseClaimContract(t *testing.T) {
if started.Task.Status != string(domain.TaskStatusRunning) ||
started.Execution.ID == "" ||
started.Execution.CurrentStep != "PREFLIGHT" ||
started.Execution.ExpiresAt.IsZero() ||
started.Execution.OrderSubmitted {
t.Fatalf("start response = %+v", started)
}
if remaining := time.Until(started.Execution.ExpiresAt); remaining < 9*time.Minute ||
remaining > 11*time.Minute {
t.Fatalf("start execution expiry remaining = %s", remaining)
}
assertNoClaimSecret(t, startedResponse, testOpaqueToken)
heartbeatBody := fmt.Sprintf(
@@ -464,6 +469,8 @@ func TestDeviceStartAndTaskHeartbeatUseClaimContract(t *testing.T) {
if heartbeat.Task.Status != string(domain.TaskStatusRunning) ||
heartbeat.Execution.ID != started.Execution.ID ||
heartbeat.Execution.CurrentStep != "SEARCH_RESULTS" ||
heartbeat.Execution.ExpiresAt.IsZero() ||
heartbeat.Execution.ExpiresAt.Before(started.Execution.ExpiresAt) ||
heartbeat.Execution.OrderSubmitted ||
heartbeat.CancelRequested {
t.Fatalf("task heartbeat response = %+v", heartbeat)
@@ -918,9 +925,10 @@ type deviceLifecycleResponse struct {
ReferenceImageURL string `json:"reference_image_url"`
} `json:"task"`
Execution struct {
ID string `json:"id"`
CurrentStep string `json:"current_step"`
OrderSubmitted bool `json:"order_submitted"`
ID string `json:"id"`
CurrentStep string `json:"current_step"`
OrderSubmitted bool `json:"order_submitted"`
ExpiresAt time.Time `json:"execution_expires_at"`
} `json:"execution"`
Replayed bool `json:"replayed"`
CancelRequested bool `json:"cancel_requested"`