| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "encoding/hex" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | "strings" |
| 13 | "time" |
| 14 | |
| 15 | "reasonix/internal/fileutil" |
| 16 | filelock "reasonix/internal/identitylock" |
| 17 | ) |
| 18 | |
| 19 | type frozenPreview struct { |
| 20 | dir string |
| 21 | freezeDir string |
| 22 | manifestBytes []byte |
| 23 | eventPath string |
| 24 | manifest Manifest |
| 25 | source Source |
| 26 | logName string |
| 27 | } |
| 28 | |
| 29 | type PrototypeImportResult struct { |
| 30 | TargetID string |
| 31 | TargetDir string |
| 32 | Source Source |
| 33 | Reused bool |
| 34 | ImportedEvents uint64 |
| 35 | } |
| 36 | |
| 37 | // ImportPrototype performs the deliberately narrow bridge from the sidecar |
| 38 | // prototype codec to the final linear codec. It accepts only the complete, |
| 39 | // validated event prefix. Unknown required events and damaged complete records |
| 40 | // fail closed; an unterminated tail is preserved with the frozen source. |
| 41 | func ImportPrototype(ctx context.Context, sourceDir, targetRoot string) (PrototypeImportResult, error) { |
| 42 | return importPreview(ctx, sourceDir, targetRoot) |
| 43 | } |
| 44 | |
| 45 | // ImportStoredPreview uses the existing explicit adapter for pre-ownership |
| 46 | // stores, but publishes to a separate staging root. The source is never |
| 47 | // upgraded in place, including for the unpublished v4 draft. |
| 48 | func ImportStoredPreview(ctx context.Context, sourceDir, targetRoot string) (PrototypeImportResult, error) { |
| 49 | if err := ctx.Err(); err != nil { |
| 50 | return PrototypeImportResult{}, err |
| 51 | } |
| 52 | if strings.TrimSpace(targetRoot) == "" || filepath.Clean(targetRoot) == "." { |
| 53 | return PrototypeImportResult{}, errors.New("session: preview target root is required") |
| 54 | } |
| 55 | frozen, err := freezePairedPreview(ctx, sourceDir) |
| 56 | if err != nil { |
| 57 | return PrototypeImportResult{}, err |
| 58 | } |
| 59 | defer os.RemoveAll(frozen.freezeDir) |
| 60 | return importFrozenPreview(ctx, frozen, targetRoot, CreateOptions{}) |
| 61 | } |
| 62 | |
| 63 | // importPreview accepts both retired prototype codecs produced before the |
| 64 | // identity cutover. Callers must resolve it together with the paired legacy |
| 65 | // transcript; opening either source in isolation can silently drop newer work. |
| 66 | func importPreview(ctx context.Context, sourceDir, targetRoot string) (PrototypeImportResult, error) { |
| 67 | if err := ctx.Err(); err != nil { |
| 68 | return PrototypeImportResult{}, err |
| 69 | } |
| 70 | sourceDir = filepath.Clean(strings.TrimSpace(sourceDir)) |
| 71 | targetRoot = filepath.Clean(strings.TrimSpace(targetRoot)) |
| 72 | if sourceDir == "." || targetRoot == "." { |
| 73 | return PrototypeImportResult{}, fmt.Errorf("session: prototype source and target root are required") |
| 74 | } |
| 75 | frozen, err := freezePreview(ctx, sourceDir) |
| 76 | if err != nil { |
| 77 | return PrototypeImportResult{}, err |
| 78 | } |
| 79 | defer os.RemoveAll(frozen.freezeDir) |
| 80 | return importFrozenPreview(ctx, frozen, targetRoot, CreateOptions{}) |
| 81 | } |
| 82 | |
| 83 | func freezePreview(ctx context.Context, sourceDir string) (frozenPreview, error) { |
| 84 | return freezePreviewCodec(ctx, sourceDir, false) |
| 85 | } |
| 86 | |
| 87 | // freezePairedPreview also accepts the unpublished v4 draft (missing |
| 88 | // storageRevision). Only migration may interpret that layout; normal session |
| 89 | // opens require the final revision explicitly. |
| 90 | func freezePairedPreview(ctx context.Context, sourceDir string) (frozenPreview, error) { |
| 91 | return freezePreviewCodec(ctx, sourceDir, true) |
| 92 | } |
| 93 | |
| 94 | func freezePreviewCodec(ctx context.Context, sourceDir string, allowCurrent bool) (frozenPreview, error) { |
| 95 | if _, err := os.Stat(sourceDir); err != nil { |
| 96 | // Report absence before taking any lock. The ownership lock lives beside |
| 97 | // the directory, so a missing candidate must not surface as a lock error |
| 98 | // that callers cannot classify as "no paired source". |
| 99 | return frozenPreview{}, err |
| 100 | } |
| 101 | // Prepare never waits behind a live writer. After the host suspends its own |
| 102 | // producer, any remaining owner makes this import ineligible. |
| 103 | releaseDirectory, err := filelock.TryAcquireMode(directoryOwnershipPath(sourceDir), filelock.ModeShared) |
| 104 | if err != nil { |
| 105 | if errors.Is(err, filelock.ErrHeld) { |
| 106 | return frozenPreview{}, fmt.Errorf("%w: freeze preview ownership", ErrWriterOwned) |
| 107 | } |
| 108 | return frozenPreview{}, fmt.Errorf("freeze preview ownership: %w", err) |
| 109 | } |
| 110 | defer releaseDirectory() |
| 111 | info, err := os.Stat(sourceDir) |
| 112 | if err != nil { |
| 113 | return frozenPreview{}, err |
| 114 | } |
| 115 | if !info.IsDir() { |
| 116 | return frozenPreview{}, fmt.Errorf("session: preview path is not a directory: %s", sourceDir) |
| 117 | } |
| 118 | releaseWriter, err := filelock.TryAcquireMode(filepath.Join(sourceDir, "writer.lock"), filelock.ModeShared) |
| 119 | if err != nil { |
| 120 | if errors.Is(err, filelock.ErrHeld) { |
| 121 | return frozenPreview{}, fmt.Errorf("%w: freeze preview writer", ErrWriterOwned) |
| 122 | } |
| 123 | return frozenPreview{}, fmt.Errorf("freeze preview writer: %w", err) |
| 124 | } |
| 125 | defer releaseWriter() |
| 126 | manifestBytes, err := os.ReadFile(filepath.Join(sourceDir, "manifest.json")) |
| 127 | if err != nil { |
| 128 | return frozenPreview{}, err |
| 129 | } |
| 130 | var manifest Manifest |
| 131 | if err := json.Unmarshal(manifestBytes, &manifest); err != nil { |
| 132 | return frozenPreview{}, fmt.Errorf("%w: prototype manifest: %w", ErrDamagedStore, err) |
| 133 | } |
| 134 | legacyCodec := manifest.SchemaVersion == 3 && (manifest.Codec == PrototypeCodec || manifest.Codec == LegacyLinearCodec || manifest.Codec == FinalV31Codec) |
| 135 | draftCodec := allowCurrent && manifest.SchemaVersion == SchemaVersion && manifest.Codec == Codec && manifest.StorageRevision == 0 |
| 136 | if (!legacyCodec && !draftCodec) || strings.TrimSpace(manifest.SessionID) == "" { |
| 137 | return frozenPreview{}, fmt.Errorf("%w: unsupported preview codec %q", ErrUnsupportedVersion, manifest.Codec) |
| 138 | } |
| 139 | logName := legacyLogName |
| 140 | if draftCodec { |
| 141 | logName = currentLogName |
| 142 | } |
| 143 | freezeDir, err := os.MkdirTemp("", "reasonix-preview-freeze-*") |
| 144 | if err != nil { |
| 145 | return frozenPreview{}, err |
| 146 | } |
| 147 | keepFreeze := false |
| 148 | defer func() { |
| 149 | if !keepFreeze { |
| 150 | _ = os.RemoveAll(freezeDir) |
| 151 | } |
| 152 | }() |
| 153 | frozenManifestPath := filepath.Join(freezeDir, "manifest.json") |
| 154 | if err := fileutil.AtomicWriteFileStrict(frozenManifestPath, manifestBytes, 0o600); err != nil { |
| 155 | return frozenPreview{}, err |
| 156 | } |
| 157 | frozenEventPath := filepath.Join(freezeDir, logName) |
| 158 | sourceEventPath := filepath.Join(sourceDir, logName) |
| 159 | if err := copyFrozenArtifact(ctx, sourceEventPath, frozenEventPath, 0o600); os.IsNotExist(err) { |
| 160 | if err := fileutil.AtomicWriteFileStrict(frozenEventPath, nil, 0o600); err != nil { |
| 161 | return frozenPreview{}, err |
| 162 | } |
| 163 | } else if err != nil { |
| 164 | return frozenPreview{}, err |
| 165 | } |
| 166 | digest := sha256.New() |
| 167 | digest.Write(manifestBytes) |
| 168 | digest.Write([]byte{0}) |
| 169 | eventFile, err := os.Open(frozenEventPath) |
| 170 | if err != nil { |
| 171 | return frozenPreview{}, err |
| 172 | } |
| 173 | eventSize, err := copyStreamWithContext(ctx, digest, eventFile) |
| 174 | closeErr := eventFile.Close() |
| 175 | if err != nil { |
| 176 | return frozenPreview{}, err |
| 177 | } |
| 178 | if closeErr != nil { |
| 179 | return frozenPreview{}, closeErr |
| 180 | } |
| 181 | sourceDigest := hex.EncodeToString(digest.Sum(nil)) |
| 182 | source := Source{Path: sourceDir, Size: int64(len(manifestBytes)) + eventSize, SHA256: sourceDigest, Version: manifest.Codec} |
| 183 | keepFreeze = true |
| 184 | return frozenPreview{dir: sourceDir, freezeDir: freezeDir, manifestBytes: manifestBytes, eventPath: frozenEventPath, manifest: manifest, source: source, logName: logName}, nil |
| 185 | } |
| 186 | |
| 187 | func importFrozenPreview(ctx context.Context, frozen frozenPreview, targetRoot string, options CreateOptions) (PrototypeImportResult, error) { |
| 188 | prototype, source := frozen.manifest, frozen.source |
| 189 | manifestBytes, sourceDir := frozen.manifestBytes, frozen.dir |
| 190 | targetID := deterministicID("prototype-import\x00" + prototype.Codec + "\x00" + sourceDir + "\x00" + source.SHA256) |
| 191 | targetDir := filepath.Join(targetRoot, targetID) |
| 192 | result := PrototypeImportResult{TargetID: targetID, TargetDir: targetDir, Source: source} |
| 193 | reused, inherited, err := reuseFrozenPreviewTarget(targetDir, targetID, source, prototype.Codec, options) |
| 194 | if err != nil { |
| 195 | return PrototypeImportResult{}, err |
| 196 | } |
| 197 | if reused { |
| 198 | result.Reused, result.ImportedEvents = true, inherited |
| 199 | return result, nil |
| 200 | } |
| 201 | if err := os.MkdirAll(targetRoot, 0o700); err != nil { |
| 202 | return PrototypeImportResult{}, err |
| 203 | } |
| 204 | tmp, err := os.MkdirTemp(targetRoot, "."+targetID+".prototype-") |
| 205 | if err != nil { |
| 206 | return PrototypeImportResult{}, err |
| 207 | } |
| 208 | published := false |
| 209 | defer func() { |
| 210 | if !published { |
| 211 | _ = os.RemoveAll(tmp) |
| 212 | } |
| 213 | }() |
| 214 | legacyDir := filepath.Join(tmp, "legacy", "prototype") |
| 215 | if err := os.MkdirAll(legacyDir, 0o700); err != nil { |
| 216 | return PrototypeImportResult{}, err |
| 217 | } |
| 218 | if err := fileutil.AtomicWriteFileStrict(filepath.Join(legacyDir, "manifest.json"), manifestBytes, 0o600); err != nil { |
| 219 | return PrototypeImportResult{}, err |
| 220 | } |
| 221 | if err := copyFrozenArtifact(ctx, frozen.eventPath, filepath.Join(legacyDir, frozen.logName), 0o600); err != nil { |
| 222 | return PrototypeImportResult{}, err |
| 223 | } |
| 224 | |
| 225 | frozenLog, err := os.Open(frozen.eventPath) |
| 226 | if err != nil { |
| 227 | return PrototypeImportResult{}, err |
| 228 | } |
| 229 | finalLogPath := filepath.Join(tmp, currentLogName) |
| 230 | finalLog, err := os.OpenFile(finalLogPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) |
| 231 | if err != nil { |
| 232 | _ = frozenLog.Close() |
| 233 | return PrototypeImportResult{}, err |
| 234 | } |
| 235 | content := contentStoreForSessionDir(tmp) |
| 236 | projection := Projection{} |
| 237 | var importedCommits int |
| 238 | var lastSequence uint64 |
| 239 | var convertErr error |
| 240 | pending := make([]Commit, 0, 64) |
| 241 | flushPending := func() error { |
| 242 | if len(pending) == 0 { |
| 243 | return nil |
| 244 | } |
| 245 | _, err := encodeV4Commits(ctx, finalLog, content, pending) |
| 246 | pending = pending[:0] |
| 247 | return err |
| 248 | } |
| 249 | knownKinds := ProjectionKinds |
| 250 | if prototype.Codec == PrototypeCodec { |
| 251 | knownKinds = PrototypeProjectionKinds |
| 252 | } |
| 253 | visit := func(_ int64, commit Commit) bool { |
| 254 | converted, err := convertPrototypeCommit(commit, prototype.SessionID, targetID, prototype.Codec) |
| 255 | if err == nil { |
| 256 | err = applyProjectionCommit(&projection, converted) |
| 257 | } |
| 258 | if err == nil { |
| 259 | pending = append(pending, converted) |
| 260 | if len(pending) == cap(pending) { |
| 261 | err = flushPending() |
| 262 | } |
| 263 | } |
| 264 | if err != nil { |
| 265 | convertErr = err |
| 266 | return false |
| 267 | } |
| 268 | importedCommits++ |
| 269 | lastSequence = converted.LastSequence() |
| 270 | return ctx.Err() == nil |
| 271 | } |
| 272 | if prototype.Codec == Codec && prototype.StorageRevision == 0 { |
| 273 | err = scanV4CommitFile(ctx, frozenLog, 0, 1, contentStoreForSessionDir(sourceDir), knownKinds, visit) |
| 274 | } else { |
| 275 | err = scanCommitFileCodec(frozenLog, 0, 1, prototype.Codec, knownKinds, visit) |
| 276 | } |
| 277 | frozenCloseErr := frozenLog.Close() |
| 278 | if err != nil { |
| 279 | _ = finalLog.Close() |
| 280 | return PrototypeImportResult{}, err |
| 281 | } |
| 282 | if convertErr != nil { |
| 283 | _ = finalLog.Close() |
| 284 | return PrototypeImportResult{}, convertErr |
| 285 | } |
| 286 | if err := flushPending(); err != nil { |
| 287 | _ = finalLog.Close() |
| 288 | return PrototypeImportResult{}, err |
| 289 | } |
| 290 | if err := ctx.Err(); err != nil { |
| 291 | _ = finalLog.Close() |
| 292 | return PrototypeImportResult{}, err |
| 293 | } |
| 294 | if frozenCloseErr != nil { |
| 295 | _ = finalLog.Close() |
| 296 | return PrototypeImportResult{}, frozenCloseErr |
| 297 | } |
| 298 | if err := finalLog.Sync(); err != nil { |
| 299 | _ = finalLog.Close() |
| 300 | return PrototypeImportResult{}, err |
| 301 | } |
| 302 | if err := finalLog.Close(); err != nil { |
| 303 | return PrototypeImportResult{}, err |
| 304 | } |
| 305 | finalManifest := Manifest{ |
| 306 | SchemaVersion: SchemaVersion, Codec: Codec, StorageRevision: StorageRevision, ContentRoot: sharedContentRoot, SessionID: targetID, |
| 307 | CreatedAt: time.Now().UTC(), InheritedEvents: lastSequence, Source: &source, |
| 308 | } |
| 309 | if err := writeImportedPreviewManifest(tmp, finalManifest, options); err != nil { |
| 310 | return PrototypeImportResult{}, err |
| 311 | } |
| 312 | validatedLog, err := os.Open(finalLogPath) |
| 313 | if err != nil { |
| 314 | return PrototypeImportResult{}, err |
| 315 | } |
| 316 | validatedProjection := Projection{} |
| 317 | validatedCommits := 0 |
| 318 | var validationErr error |
| 319 | replayErr := scanV4CommitFile(ctx, validatedLog, 0, 1, content, nil, func(_ int64, commit Commit) bool { |
| 320 | if err := applyProjectionCommit(&validatedProjection, commit); err != nil { |
| 321 | validationErr = err |
| 322 | return false |
| 323 | } |
| 324 | validatedCommits++ |
| 325 | return true |
| 326 | }) |
| 327 | closeErr := validatedLog.Close() |
| 328 | replayErr = errors.Join(replayErr, validationErr, closeErr) |
| 329 | if replayErr != nil || validatedCommits != importedCommits || validatedProjection.CommittedSequence != lastSequence { |
| 330 | if replayErr == nil { |
| 331 | replayErr = fmt.Errorf("replayed %d commits through sequence %d; want %d through %d", validatedCommits, validatedProjection.CommittedSequence, importedCommits, lastSequence) |
| 332 | } |
| 333 | return PrototypeImportResult{}, fmt.Errorf("validate prototype target: %w", replayErr) |
| 334 | } |
| 335 | if err := os.Rename(tmp, targetDir); err != nil { |
| 336 | return PrototypeImportResult{}, fmt.Errorf("publish prototype target: %w", err) |
| 337 | } |
| 338 | published = true |
| 339 | result.ImportedEvents = lastSequence |
| 340 | return result, nil |
| 341 | } |
| 342 | |
| 343 | func reuseFrozenPreviewTarget(targetDir, targetID string, source Source, codec string, options CreateOptions) (bool, uint64, error) { |
| 344 | manifest, err := readManifest(filepath.Join(targetDir, "manifest.json")) |
| 345 | if os.IsNotExist(err) { |
| 346 | return false, 0, nil |
| 347 | } |
| 348 | if err != nil { |
| 349 | return false, 0, err |
| 350 | } |
| 351 | if manifest.Source == nil || manifest.Source.Path != source.Path || manifest.Source.SHA256 != source.SHA256 || manifest.Source.Version != codec { |
| 352 | return false, 0, fmt.Errorf("%w: prototype target %s has another source", ErrSessionExists, targetID) |
| 353 | } |
| 354 | if err := validateSessionHeaderForCreate(targetDir, targetID, options); err != nil { |
| 355 | return false, 0, err |
| 356 | } |
| 357 | return true, manifest.InheritedEvents, nil |
| 358 | } |
| 359 | |
| 360 | func writeImportedPreviewManifest(dir string, manifest Manifest, options CreateOptions) error { |
| 361 | if err := writeManifestFile(filepath.Join(dir, "manifest.json"), manifest); err != nil { |
| 362 | return err |
| 363 | } |
| 364 | return writeSessionHeaderForCreate(dir, manifest.SessionID, manifest.CreatedAt, options) |
| 365 | } |
| 366 | |
| 367 | func convertPrototypeCommit(original Commit, sourceID, targetID, sourceCodec string) (Commit, error) { |
| 368 | commit := cloneCommit(original) |
| 369 | commit.SchemaVersion = SchemaVersion |
| 370 | commit.Codec = Codec |
| 371 | commit.ID = deterministicID("prototype-commit\x00" + targetID + "\x00" + original.ID) |
| 372 | commit.OperationID = "prototype:" + sourceID + ":" + original.ID |
| 373 | commit.WriterGeneration = 1 |
| 374 | for eventIndex := range commit.Events { |
| 375 | if sourceCodec == PrototypeCodec && commit.Events[eventIndex].Kind == "context/replace" { |
| 376 | commit.Events[eventIndex].Kind = "history/replace" |
| 377 | } |
| 378 | } |
| 379 | operationHash, err := hashOperation(targetID, commit.TurnID, commit.Events) |
| 380 | if err != nil { |
| 381 | return Commit{}, err |
| 382 | } |
| 383 | commit.OperationHash = operationHash |
| 384 | return commit, nil |
| 385 | } |
| 386 |