package daemon

// quarantineSnapshotCoordinator prevents a deleted session from starting any
// further publication. The returned join channel is closed after an already
// started repository call returns; callers must wait without daemon or session
// locks before deleting repository state.
func quarantineSnapshotCoordinator(sess *session) <-chan struct{} {
	return quarantineSnapshotCoordinatorWithOptions(sess, false)
}

// quarantineSnapshotCoordinatorRetainingQueuedCapture is used after a forced
// shutdown checkpoint times out. Its queued capture is logical retry state, not
// a repository operation: startSnapshotPublication rejects it while
// quarantined, and the worker eventually discards it after its blocked call
// returns.
func quarantineSnapshotCoordinatorRetainingQueuedCapture(sess *session) <-chan struct{} {
	return quarantineSnapshotCoordinatorWithOptions(sess, true)
}

func quarantineSnapshotCoordinatorWithOptions(sess *session, retainQueuedCapture bool) <-chan struct{} {
	if sess == nil {
		done := make(chan struct{})
		close(done)
		return done
	}
	sess.snapshotMu.Lock()
	sess.snapshotQuarantined = true
	sess.snapEligible.Store(false)
	if sess.snapshotPublicationCancel != nil {
		sess.snapshotPublicationCancel()
	}
	if sess.snapshotQueuedCapture != nil && !retainQueuedCapture {
		// A rename or purge must discard a capture under the old identity. The
		// global queue may still contain it, but startSnapshotPublication
		// compares identity and will discard it even if this session is renamed.
		sess.snapshotQueuedCapture.coordinatorDiscarded = true
		sess.snapshotQueuedCapture = nil
		if sess.snapshotPendingCaptures > 0 {
			sess.snapshotPendingCaptures--
		}
		sess.snapshotPending = sess.snapshotPendingCaptures > 0 || sess.snapshotInFlightCapture != nil
	}
	done := sess.snapshotPublicationDone
	if done == nil {
		done = make(chan struct{})
		close(done)
	}
	sess.signalSnapshotChangedLocked()
	sess.snapshotMu.Unlock()
	return done
}

// rollbackSnapshotQuarantine reverses quarantineSnapshotCoordinator when a
// teardown aborts before it owns any destructive repository operation and the
// session therefore stays live and registered. The quarantine cancelled the
// publication context and flipped eligibility, so a surviving session would
// otherwise never checkpoint again: this discards the cancelled context so the
// next schedule allocates a fresh one, restores the captured pre-quarantine
// eligibility, and marks the session dirty so the mutations its ineligible fast
// path suppressed during the quarantine window are captured by a later
// publication. A queued capture the quarantine discarded stays discarded:
// startSnapshotPublication rejects it by identity and the forced dirty
// generation guarantees a replacement capture.
func rollbackSnapshotQuarantine(sess *session, wasEligible bool) {
	if sess == nil {
		return
	}
	sess.snapshotMu.Lock()
	sess.snapshotQuarantined = false
	sess.snapshotPublicationContext = nil
	sess.snapshotPublicationCancel = nil
	sess.snapshotMu.Unlock()
	sess.snapEligible.Store(wasEligible)
	if wasEligible {
		markSnapshotDirty(sess)
	}
}

func markSnapshotDirty(sess *session) {
	if sess == nil || !sess.snapEligible.Load() {
		return
	}
	sess.snapshotMu.Lock()
	sess.snapshotGeneration++
	sess.snapDirty.Store(true)
	sess.signalSnapshotChangedLocked()
	wake := sess.snapshotWake
	sess.snapshotMu.Unlock()
	if wake != nil {
		select {
		case wake <- struct{}{}:
		default:
		}
	}
}

// snapshotChangeLocked returns a channel closed by the next snapshot state
// transition. snapshotMu must be held.
func (sess *session) snapshotChangeLocked() chan struct{} {
	if sess.snapshotChanged == nil {
		sess.snapshotChanged = make(chan struct{})
	}
	return sess.snapshotChanged
}

// signalSnapshotChangedLocked wakes waiters after a snapshot state transition.
// snapshotMu must be held.
func (sess *session) signalSnapshotChangedLocked() {
	close(sess.snapshotChangeLocked())
	sess.snapshotChanged = make(chan struct{})
}

// waitForSnapshotGeneration waits only for the bounded terminal checkpoint
// deadline. It observes successful publication of the requested mutation
// revision, rather than dirty state, so a concurrent later mutation cannot
// turn a completed forced checkpoint into a false timeout.
func (d *Daemon) waitForSnapshotGeneration(sess *session, generation uint64) bool {
	if d == nil || sess == nil || generation == 0 {
		return true
	}
	deadline := newSnapshotShutdownDeadline(d.clock)
	defer deadline.stop()
	return d.waitForSnapshotGenerationWithDeadline(sess, generation, deadline)
}

// waitForSnapshotGenerationWithDeadline waits for a forced checkpoint without
// allocating another shutdown interval. Serve shares one deadline across every
// session and the subsequent worker drain.
func (d *Daemon) waitForSnapshotGenerationWithDeadline(sess *session, generation uint64, deadline *snapshotShutdownDeadline) bool {
	if d == nil || sess == nil || generation == 0 {
		return true
	}
	if deadline == nil {
		return d.waitForSnapshotGeneration(sess, generation)
	}
	for {
		// Once the shared deadline is observable, timeout wins even if publication
		// completion becomes ready concurrently. This preserves the timed-out
		// session identity and avoids nondeterministic success after the budget.
		select {
		case <-deadline.Done():
			return false
		default:
		}
		sess.snapshotMu.Lock()
		published := sess.snapshotPublishedMutationRevision >= generation
		changed := sess.snapshotChangeLocked()
		sess.snapshotMu.Unlock()
		if published {
			return true
		}
		select {
		case <-changed:
		case <-deadline.Done():
			return false
		}
	}
}

// startSnapshotPublication moves a retained capture from the producer queue to
// the repository worker. Quarantine and this transition share snapshotMu, so a
// repository call cannot begin after destructive teardown has quarantined the
// session.
func startSnapshotPublication(capture *snapshotCapture) bool {
	if capture == nil || capture.session == nil {
		return false
	}
	sess := capture.session
	sess.snapshotMu.Lock()
	defer sess.snapshotMu.Unlock()
	if sess.snapshotQuarantined || sess.snapshotQueuedCapture != capture {
		return false
	}
	sess.snapshotQueuedCapture = nil
	sess.snapshotInFlightCapture = capture
	sess.snapshotPublicationDone = make(chan struct{})
	sess.snapshotPending = true
	sess.signalSnapshotChangedLocked()
	return true
}

func (d *Daemon) finishSnapshotCapture(capture *snapshotCapture, succeeded bool) {
	if capture == nil {
		return
	}
	capture.finishOnce.Do(func() {
		var shouldScheduleForcedSuccessor bool
		capture.session.snapshotMu.Lock()
		// A capture the quarantine discarded already dropped its pending count when
		// it was detached from the producer queue; the worker still calls finish on
		// it after rejecting the stale identity. Decrementing again here would
		// under-count a replacement capture scheduled after a rollback, so a
		// discarded capture must not decrement twice.
		if !capture.coordinatorDiscarded && capture.session.snapshotPendingCaptures > 0 {
			capture.session.snapshotPendingCaptures--
		}
		if capture.session.snapshotQueuedCapture == capture {
			capture.session.snapshotQueuedCapture = nil
		}
		if capture.session.snapshotInFlightCapture == capture {
			capture.session.snapshotInFlightCapture = nil
			if capture.session.snapshotPublicationDone != nil {
				close(capture.session.snapshotPublicationDone)
				capture.session.snapshotPublicationDone = nil
			}
		}
		capture.session.snapshotPending = capture.session.snapshotPendingCaptures > 0 ||
			capture.session.snapshotQueuedCapture != nil || capture.session.snapshotInFlightCapture != nil
		if !capture.coordinatorDiscarded && capture.attemptKind == snapshotAttemptRoutine && d.snapshotRepository != nil {
			// The interval is measured from completion, including a failed
			// attempt. A forced teardown checkpoint deliberately leaves it alone.
			capture.session.snapshotNextEligibleAt = d.clock.Now().Add(snapshotInterval)
		}
		mutationRevision := capture.mutationRevision
		// Older unit fixtures created captures before mutation revisions were
		// recorded explicitly. Production captures always set it at scheduling;
		// retaining this fallback keeps those fixtures representative.
		if mutationRevision == 0 {
			mutationRevision = capture.generation
		}
		if succeeded {
			if capture.generation >= capture.session.snapshotPublishedGeneration &&
				capture.checkpoint.Generation == capture.generation && capture.checkpoint.ManifestDigest != ([32]byte{}) {
				checkpoint := capture.checkpoint
				capture.session.snapshotPublishedCheckpoint = &checkpoint
			}
			capture.session.snapshotPublishedGeneration = max(capture.session.snapshotPublishedGeneration, capture.generation)
			capture.session.snapshotPublishedMutationRevision = max(capture.session.snapshotPublishedMutationRevision, mutationRevision)
		}
		if succeeded && capture.session.snapshotGeneration == mutationRevision {
			capture.session.snapDirty.Store(false)
		} else if !succeeded {
			capture.session.snapDirty.Store(true)
		}
		if succeeded && capture.session.snapshotForcedGeneration != 0 &&
			mutationRevision >= capture.session.snapshotForcedGeneration {
			capture.session.snapshotForcedGeneration = 0
		}
		// A failed forced publication remains dirty for ordinary retry rather
		// than recursively creating forced jobs. A later forced request records
		// a newer intent and can create its one bounded successor.
		shouldScheduleForcedSuccessor = capture.session.snapshotForcedGeneration != 0 &&
			mutationRevision < capture.session.snapshotForcedGeneration &&
			capture.session.snapshotQueuedCapture == nil && capture.session.snapshotInFlightCapture == nil &&
			(succeeded || capture.attemptKind != snapshotAttemptForced)
		capture.session.signalSnapshotChangedLocked()
		wake := capture.session.snapshotWake
		capture.session.snapshotMu.Unlock()
		if capture.normalWorkerAdmitted {
			d.snapshots.mu.Lock()
			delete(d.snapshots.admitted, capture)
			d.snapshots.mu.Unlock()
		}
		if wake != nil {
			select {
			case wake <- struct{}{}:
			default:
			}
		}
		if shouldScheduleForcedSuccessor {
			// This runs after publication completion and outside snapshotMu. It
			// bypasses routine eligibility while preserving the single queued and
			// single in-flight capture bound.
			_ = d.scheduleFinalSnapshot(capture.session)
		}
		if succeeded {
			d.clearSnapshotFailure()
		}
	})
}