姚章浩
8d838d9cb4
build / build (push) Successful in 1m45s
执行时真实时钟与持久授权回读（撤销/过期即停并保留预备份）、写集合写前落盘、 租约与 attempt 身份、恢复队列与原发送身份分离、成功收尾必须释放维护后再登记终态。
69 lines
2.6 KiB
Go
69 lines
2.6 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"knowledge-governance/internal/governance"
|
|
"knowledge-governance/internal/jobs"
|
|
"time"
|
|
)
|
|
|
|
func (s *Store) CreateLibraryDispatch(ctx context.Context, job governance.LibraryJob, stages []string, handoff bool, now time.Time) (governance.LibraryJob, bool, error) {
|
|
tx, err := s.db.Begin(ctx)
|
|
if err != nil {
|
|
return governance.LibraryJob{}, false, err
|
|
}
|
|
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
|
|
created, reused, err := NewStore(tx).WithLibraryTarget(s.libraryTargetID).createLibraryDispatch(ctx, job, stages, handoff, now)
|
|
if err != nil {
|
|
return governance.LibraryJob{}, false, err
|
|
}
|
|
if err = tx.Commit(ctx); err != nil {
|
|
return governance.LibraryJob{}, false, err
|
|
}
|
|
return created, reused, nil
|
|
}
|
|
func (s *Store) createLibraryDispatch(ctx context.Context, job governance.LibraryJob, stages []string, handoff bool, now time.Time) (governance.LibraryJob, bool, error) {
|
|
created, reused, err := s.CreateLibraryJob(ctx, job, stages, now)
|
|
if err != nil {
|
|
return created, reused, err
|
|
}
|
|
if reused && (created.State != governance.LibraryJobPending || created.QueueJobID != "") {
|
|
return created, true, nil
|
|
}
|
|
kind := "library." + string(created.Kind)
|
|
if created.Kind != governance.LibraryJobSync && created.Kind != governance.LibraryJobRestore && created.Kind != governance.LibraryJobBaseline {
|
|
return created, reused, errors.New("unsupported library dispatch kind")
|
|
}
|
|
queueID, kind, raw, err := governance.LibraryQueueRequest(handoff, created.ID, created.Scope, created.TargetSnapshotID, kind)
|
|
if err != nil {
|
|
return created, reused, err
|
|
}
|
|
var payload map[string]string
|
|
if err = json.Unmarshal(raw, &payload); err != nil {
|
|
return created, reused, err
|
|
}
|
|
queued, err := s.Enqueue(ctx, jobs.EnqueueRequest{ID: queueID, Type: kind, Payload: raw, AvailableAt: now, MaxAttempts: 1, IdempotencyKey: kind + ".queue:" + created.ID}, now)
|
|
if err != nil {
|
|
return created, reused, err
|
|
}
|
|
if queued == nil || queued.ID != queueID || queued.Type != kind {
|
|
return created, reused, errors.New("queue identity conflict")
|
|
}
|
|
var actual map[string]string
|
|
if json.Unmarshal(queued.Payload, &actual) != nil || len(actual) != len(payload) {
|
|
return created, reused, errors.New("queue payload mismatch")
|
|
}
|
|
for key, value := range payload {
|
|
if actual[key] != value {
|
|
return created, reused, errors.New("queue payload mismatch")
|
|
}
|
|
}
|
|
if err := s.AttachLibraryQueueJob(ctx, created.ID, queued.ID, now); err != nil {
|
|
return created, reused, err
|
|
}
|
|
created.QueueJobID = queued.ID
|
|
return created, reused, nil
|
|
}
|