knowledge-governance/cmd/worker/library_completion.go

package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"knowledge-governance/internal/governance"
	"knowledge-governance/internal/libraryrelease"
	postgresstore "knowledge-governance/internal/platform/postgres"
)

// libraryFailStore 是"失败出口"需要的写面：带代次守卫的失败收尾。
// 抽成接口是因为 writer 与裸 Store 都要能传进来（测试里常用裸 Store）。
type libraryFailStore interface {
	AdvanceLibraryJob(ctx context.Context, jobID string, state governance.LibraryJobState, stage, message, recovery, snapshotID, targetSnapshotID string, now time.Time) error
	UpdateLibraryMaintenanceState(ctx context.Context, scope, ownerJobID, state, note string, now time.Time) error
}

// prebackupAnchors 读"这次操作第一次拍下的那份额度"：字节数与写集合摘要。
//
// 为什么必须读它而不是重新计算：接管/收尾续作都没有重新备份，手上只有已登记的事实；
// 重新用当前状态算一遍就等于把"这次用了多大一份基线、写了什么"换成猜的（F09 的现场
// 就是 backup_bytes 变成 0、writeset_digest 变空）。
func prebackupAnchors(ctx context.Context, store *postgresstore.Store, scope, jobID, backupSnapshot string) (map[string]any, error) {
	version, found, err := store.LibraryVersionByRef(ctx, scope, governance.LibraryVersionProductionPrebackup, jobID, backupSnapshot)
	if err != nil {
		return nil, fmt.Errorf("读不到预备份证据：%w", err)
	}
	if !found {
		return nil, fmt.Errorf("预备份 %s 没有作业 %s 的证据行：拒绝用空锚点登记生效版本", backupSnapshot, jobID)
	}
	anchors := map[string]any{}
	for _, key := range []string{"backup_bytes", "writeset_digest", "writeset_rows", "writeset_path",
		"writeset_created_at", "backup_created_at", "index_mapping_sha256", "backup_reused"} {
		if value, ok := version.Evidence[key]; ok && value != nil {
			anchors[key] = value
		}
	}
	return anchors, nil
}

// finalizeEvidence 构造"生效/恢复"事件的证据：以持久化锚点为底，补上本次动作与映射。
// 缺字节数或写集合摘要时返回错误——收尾可以补登记，但不能补出一份没有锚点的"成功"。
func finalizeEvidence(ctx context.Context, store *postgresstore.Store, scope, jobID, backupSnapshot,
	targetSnapshot, action, mappingDigest string) (map[string]any, error) {
	anchors, err := prebackupAnchors(ctx, store, scope, jobID, backupSnapshot)
	if err != nil {
		return nil, err
	}
	if _, ok := anchors["backup_bytes"]; !ok {
		return nil, fmt.Errorf("预备份 %s 的证据里没有真实字节数：拒绝登记生效版本（会把 backup_bytes 洗成空）", backupSnapshot)
	}
	if digest, _ := anchors["writeset_digest"].(string); digest == "" {
		return nil, fmt.Errorf("预备份 %s 的证据里没有写集合摘要：拒绝登记生效版本（恢复将无从按计划清理）", backupSnapshot)
	}
	evidence := map[string]any{"backup_snapshot": backupSnapshot, "action": action, "target_snapshot": targetSnapshot}
	for key, value := range anchors {
		evidence[key] = value
	}
	if mappingDigest != "" {
		evidence["index_mapping_sha256"] = mappingDigest
	}
	return evidence, nil
}

// rebuildSuccessBoundary 从**持久事实**重建"内容已验证、只差收尾"的凭据。
//
// 触发场景（F05 路径一）：journal 已 saved succeeded，但进程在 finishLibrarySuccess 之前
// 崩了。接管方再跑 RunWindow 会直接成功返回、不再执行 work；如果这里还依赖进程内
// pending 指针，作业就永远收不了尾（锁挂着、队列已完成、页面一直显示维护中）。
//
// 重建需要且只需要三件事：
//  1. 作业行里持久化的预备份身份（这次用的是哪份基线）；
//  2. 那份预备份的证据行里有真实字节数与写集合摘要（首次锚点没被洗掉）；
//  3. 生效事件存在，或可以被**幂等补登记**（不重写目标、不再 Apply）。
func rebuildSuccessBoundary(ctx context.Context, store *libraryAttemptWriter, recorder libraryStageRecorder,
	payload libraryReleasePayload, previous governance.LibraryJob, kind governance.LibraryJobKind,
	grant governance.LibraryReleaseAuthorization) (successBoundary, error) {
	backup := previous.SnapshotID
	if backup == "" {
		return successBoundary{}, errors.New("窗口已成功，但作业行没有持久化预备份身份：无法证明内容基线")
	}
	action := string(kind)
	if payload.FinalizeOnly {
		action = "finalize_only"
	}
	if _, err := prebackupAnchors(ctx, store.Store, payload.Scope, payload.LibraryJobID, backup); err != nil {
		return successBoundary{}, err
	}
	_, found, err := store.LibraryVersionByRef(ctx, payload.Scope, governance.LibraryVersionProductionEffective,
		payload.LibraryJobID, payload.TargetSnapshotID)
	if err != nil {
		return successBoundary{}, fmt.Errorf("读不到生效事件：%w", err)
	}
	if !found {
		recorder.Stage(ctx, "finalize", "running", "窗口已成功：补登记生效版本（不重写生产）")
		evidence, evidenceErr := finalizeEvidence(ctx, store.Store, payload.Scope, payload.LibraryJobID, backup,
			payload.TargetSnapshotID, action, recorder.mappingDigest)
		if evidenceErr != nil {
			return successBoundary{}, evidenceErr
		}
		if err := store.RecordLibraryVersion(ctx, governance.LibraryVersion{
			ID:    governance.LibraryVersionEventID(governance.LibraryVersionProductionEffective, payload.Scope, payload.TargetSnapshotID, store.attempt.OperationID),
			Scope: payload.Scope, Kind: governance.LibraryVersionProductionEffective,
			SnapshotID: payload.TargetSnapshotID, SourceJobID: payload.LibraryJobID, CreatedBy: grant.AuthorizedBy,
			Evidence: evidence, CreatedAt: time.Now().UTC(),
		}); err != nil {
			return successBoundary{}, fmt.Errorf("补登记生效版本失败：%w", err)
		}
		recorder.Stage(ctx, "finalize", "succeeded", "已补登记生效版本（未重写生产）")
	}
	return successBoundary{Ready: true, Backup: backup, Target: payload.TargetSnapshotID,
		FinishedAt: time.Now().UTC()}, nil
}

// recordRecoveryEvent 是"按预备份恢复成功"的持久事实（F06）。
//
// 它必须回答两件事，缺一不可：
//   - 恢复出来的内容身份是什么（backup 快照 + 校验摘要）；
//   - current 现在指向它——旧实现只释放维护锁，于是物理内容已经是旧版，
//     overview/diff/provenance 仍然宣称失败的新版本生效，后续操作拿错误基线做判断。
//
// 它是独立事件，不改写原来那条失败作业/失败版本；原失败事实保持不变。
func recordRecoveryEvent(ctx context.Context, store *libraryAttemptWriter, operation governance.LibraryOperation,
	scope, jobID, backupSnapshot, verifyDigest, note string) error {
	if backupSnapshot == "" || verifyDigest == "" {
		return errors.New("恢复事件缺少预备份身份或校验摘要：不能登记成恢复事实")
	}
	// 事件挂在**恢复意图**上（不是失败作业的旧意图）：current 的次序按意图创建时间，
	// 旧意图的迟到事实推不动当前状态。
	operationID := operation.ID
	evidence := map[string]any{
		"backup_snapshot": backupSnapshot, "action": "recover", "verify_digest": verifyDigest,
		"window_job": jobID, "note": note, "recovered_at": time.Now().UTC(),
	}
	// 首次锚点若存在，一并带上：恢复事件同样要能回答"这份基线多大、写集合是什么"。
	if anchors, err := prebackupAnchors(ctx, store.Store, scope, jobID, backupSnapshot); err == nil {
		for key, value := range anchors {
			if _, exists := evidence[key]; !exists {
				evidence[key] = value
			}
		}
	}
	return store.RecordLibraryVersion(ctx, governance.LibraryVersion{
		ID: governance.LibraryVersionEventID(governance.LibraryVersionRecoveryVerified, scope, backupSnapshot,
			operationID),
		Scope: scope, Kind: governance.LibraryVersionRecoveryVerified, SnapshotID: backupSnapshot,
		SourceJobID: jobID, OperationID: operationID, AttemptEpoch: store.attempt.Epoch,
		CreatedBy: "worker", Evidence: evidence, CreatedAt: operation.CreatedAt,
	})
}

var _ libraryrelease.Recorder = libraryStageRecorder{}