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

feat(t214): persist candidate observation identities

Browse Source
This commit is contained in:
QiuSW
2026-07-28 12:21:55 +08:00
parent acd3c85b5b
commit 56e7a2be33
29 changed files with 622 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-api/internal/repository/sqlite/auth_repository_test.go
+6 -3
View File
@@ -383,6 +383,9 @@ func TestAuthMigrationCanRollbackWithoutRebuildingPurchaseTasks(
if err != nil {
t.Fatalf("migration.New() error = %v", err)
}
if err := runner.Down(context.Background()); err != nil {
t.Fatalf("Down(v7) error = %v", err)
}
if err := runner.Down(context.Background()); err != nil {
t.Fatalf("Down(v6) error = %v", err)
}
@@ -408,9 +411,9 @@ func TestAuthMigrationCanRollbackWithoutRebuildingPurchaseTasks(
t.Fatal("purchase_tasks was lost during auth migration rollback")
}
if applied, err := runner.Up(context.Background()); err != nil {
t.Fatalf("Up(v3-v6) error = %v", err)
} else if applied != 4 {
t.Fatalf("Up(v3-v6) applied = %d, want 4", applied)
t.Fatalf("Up(v3-v7) error = %v", err)
} else if applied != 5 {
t.Fatalf("Up(v3-v7) applied = %d, want 5", applied)
}
}
backend-api/internal/repository/sqlite/candidate_decision_repository.go
+80 -6
View File
@@ -7,6 +7,7 @@ import (
"encoding/hex"
"encoding/json"
"errors"
"strconv"
"cmroubao/backend-api/internal/domain"
"cmroubao/backend-api/internal/usecase"
@@ -93,6 +94,25 @@ func storeCandidateDecisionDataset(
if err != nil {
return repositoryFailure(err)
}
_, err = tx.ExecContext(
ctx,
`INSERT INTO candidate_observation_identities (
candidate_key, execution_id, candidate_ordinal,
card_signature, detail_signature, detail_evidence_sha256,
specification_evidence_sha256, identity_version, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, 1, ?)`,
candidateObservationKey(write.ExecutionID, candidate),
write.ExecutionID,
candidate.Ordinal,
candidate.CardSignature,
candidate.DetailSignature,
candidate.DetailEvidenceSHA256,
candidate.SpecificationEvidenceSHA256,
formatTimestamp(write.Now),
)
if err != nil {
return repositoryFailure(err)
}
}
if batch.ProvenanceJSON != nil {
var provenance usecase.ExecutionProvenance
@@ -567,12 +587,21 @@ func getCandidateDecisionDataset(
}
rows, err := queryer.QueryContext(
ctx,
`SELECT task_id, execution_id, ordinal, title, sku_text, price_text,
product_url, image_url, evidence_asset_ids_json,
collection_status, observed_at
FROM candidate_observations
WHERE task_id = ? AND execution_id = ?
ORDER BY ordinal ASC`,
`SELECT observation.task_id, observation.execution_id,
observation.ordinal, observation.title, observation.sku_text,
observation.price_text, observation.product_url,
observation.image_url, observation.evidence_asset_ids_json,
observation.collection_status, observation.observed_at,
identity.candidate_key, identity.card_signature,
identity.detail_signature, identity.detail_evidence_sha256,
identity.specification_evidence_sha256,
identity.identity_version, identity.created_at
FROM candidate_observations AS observation
LEFT JOIN candidate_observation_identities AS identity
ON identity.execution_id = observation.execution_id
AND identity.candidate_ordinal = observation.ordinal
WHERE observation.task_id = ? AND observation.execution_id = ?
ORDER BY observation.ordinal ASC`,
taskID,
executionID,
)
@@ -582,6 +611,9 @@ func getCandidateDecisionDataset(
for rows.Next() {
var observation domain.CandidateObservation
var evidenceJSON, observedAt string
var candidateKey, cardSignature, detailSignature sql.NullString
var detailHash, specificationHash, identityAt sql.NullString
var identityVersion sql.NullInt64
if err := rows.Scan(
&observation.TaskID,
&observation.ExecutionID,
@@ -594,6 +626,13 @@ func getCandidateDecisionDataset(
&evidenceJSON,
&observation.CollectionStatus,
&observedAt,
&candidateKey,
&cardSignature,
&detailSignature,
&detailHash,
&specificationHash,
&identityVersion,
&identityAt,
); err != nil {
_ = rows.Close()
return nil, repositoryFailure(err)
@@ -610,6 +649,24 @@ func getCandidateDecisionDataset(
_ = rows.Close()
return nil, repositoryFailure(err)
}
if candidateKey.Valid {
identityCreatedAt, parseErr := parseTimestamp(identityAt.String)
if parseErr != nil {
_ = rows.Close()
return nil, repositoryFailure(parseErr)
}
observation.Identity = &domain.CandidateObservationIdentity{
CandidateKey: candidateKey.String,
ExecutionID: observation.ExecutionID,
CandidateOrdinal: observation.Ordinal,
CardSignature: cardSignature.String,
DetailSignature: detailSignature.String,
DetailEvidenceSHA256: detailHash.String,
SpecificationEvidenceSHA256: specificationHash.String,
IdentityVersion: int(identityVersion.Int64),
CreatedAt: identityCreatedAt,
}
}
dataset.Observations = append(dataset.Observations, observation)
}
if err := rows.Err(); err != nil {
@@ -823,6 +880,23 @@ func hashCandidateResult(candidates string, recommendation *string) string {
return hex.EncodeToString(digest.Sum(nil))
}
func candidateObservationKey(
executionID string,
candidate usecase.ExecutionCandidate,
) string {
digest := sha256.New()
_, _ = digest.Write([]byte("cmroubao-candidate-v1"))
_, _ = digest.Write([]byte{0})
_, _ = digest.Write([]byte(executionID))
_, _ = digest.Write([]byte{0})
_, _ = digest.Write([]byte(strconv.Itoa(candidate.Ordinal)))
_, _ = digest.Write([]byte{0})
_, _ = digest.Write([]byte(candidate.DetailSignature))
_, _ = digest.Write([]byte{0})
_, _ = digest.Write([]byte(candidate.SpecificationEvidenceSHA256))
return hex.EncodeToString(digest.Sum(nil))
}
func nullableSQLString(value sql.NullString) any {
if !value.Valid {
return nil
backend-api/internal/repository/sqlite/candidate_decision_repository_test.go
+34
View File
@@ -0,0 +1,34 @@
package sqlite
import (
"strings"
"testing"
"cmroubao/backend-api/internal/usecase"
)
func TestCandidateObservationKeyIsStableAndObservationScoped(t *testing.T) {
candidate := usecase.ExecutionCandidate{
Ordinal: 1,
DetailSignature: strings.Repeat("c", 64),
SpecificationEvidenceSHA256: strings.Repeat("d", 64),
}
first := candidateObservationKey("execution-1", candidate)
replayed := candidateObservationKey("execution-1", candidate)
otherOrdinal := candidate
otherOrdinal.Ordinal = 2
if len(first) != 64 {
t.Fatalf("candidate key length = %d", len(first))
}
if replayed != first {
t.Fatalf("replayed key = %q, want %q", replayed, first)
}
if candidateObservationKey("execution-1", otherOrdinal) == first {
t.Fatal("different observation ordinal produced the same key")
}
if candidateObservationKey("execution-2", candidate) == first {
t.Fatal("different execution produced the same key")
}
}
backend-api/internal/repository/sqlite/execution_result_repository.go
+25 -3
View File
@@ -557,9 +557,7 @@ func validateCandidateEvidence(
write usecase.ExecutionResultWrite,
candidatesJSON string,
) error {
var candidates []struct {
EvidenceAssetIDs []string `json:"evidence_asset_ids"`
}
var candidates []usecase.ExecutionCandidate
if err := json.Unmarshal([]byte(candidatesJSON), &candidates); err != nil {
return usecase.ErrRepositoryInvariant
}
@@ -569,6 +567,30 @@ func validateCandidateEvidence(
); err != nil {
return err
}
if len(candidate.EvidenceAssetIDs) != 2 {
return usecase.ErrTaskStateConflict
}
expectedHashes := []string{
candidate.DetailEvidenceSHA256,
candidate.SpecificationEvidenceSHA256,
}
for index, evidenceID := range candidate.EvidenceAssetIDs {
var actualHash string
err := tx.QueryRowContext(
ctx,
`SELECT sha256 FROM execution_evidence_assets
WHERE id = ? AND task_id = ? AND execution_id = ?`,
evidenceID,
write.TaskID,
write.ExecutionID,
).Scan(&actualHash)
if err != nil {
return repositoryFailure(err)
}
if actualHash != expectedHashes[index] {
return usecase.ErrTaskStateConflict
}
}
}
return nil
}