knowledge-governance/internal/platform/postgres/library_dispatch.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
}