<?php
/**
 * Sync worker tick driver.
 *
 * Phase 2 (post-MVP) Application-layer collaborator that owns the
 * §11 / ADR-0021 §8 worker loop: pull the next runnable job off
 * `affilun_sync_jobs` via {@see SyncJobRepository::claimNext()}, hand it
 * to {@see SyncProductJobHandler::handle()}, and repeat up to a
 * configurable batch size before yielding control back to the
 * caller. The worker is deliberately a single-tick driver (NOT a
 * long-running process): one call to {@see self::tick()} drains at
 * most `$batchSize` jobs and returns, so the surrounding scheduler
 * (WP-Cron, CLI, integration test) controls cadence and concurrency.
 *
 * Responsibilities (per `specifications/sync-and-cache-strategy.md`
 * §11.3 / §11.4 and ADR-0021 §8.1–§8.2):
 *
 *   1. Stamp a single "now" instant from the injected {@see Clock}
 *      at the top of the tick. The same instant is forwarded to
 *      every {@see SyncJobRepository::claimNext()} call inside the
 *      tick so a slow handler cannot drift the per-claim timestamp
 *      forward and silently widen the eligibility window.
 *
 *   2. Compute the lock-timeout cutoff once at the top of the tick
 *      as `now - lockTimeoutSeconds`. Jobs whose `locked_at` is
 *      strictly older than this cutoff are reclaimed per §8.5.
 *
 *   3. Loop up to {@see self::$batchSize} times. On each iteration:
 *
 *        a. Call `claimNext($workerId, $now, $cutoff)`.
 *        b. If null, the queue is idle; break out of the loop.
 *        c. Otherwise, dispatch the claimed record to
 *           {@see SyncProductJobHandler::handle()} and accumulate
 *           the resulting {@see SyncJobOutcome} on the tick result.
 *
 *   4. Return a {@see SyncWorkerTickResult} carrying the number of
 *      jobs actually claimed plus the per-job outcomes the caller
 *      can fold into observability surfaces (logs, audit events,
 *      diagnostics counters).
 *
 * Failure semantics:
 *
 *   - The worker treats a {@see Throwable} bubbling out of
 *     {@see SyncProductJobHandler::handle()} as a fatal protocol
 *     violation: the handler contract guarantees that every error
 *     path is encoded into the returned outcome, so an exception
 *     means the handler itself crashed (programming error, missing
 *     dependency, repository unavailable). The worker re-raises
 *     the throwable so the surrounding scheduler can record the
 *     incident and back off; the in-flight job is left in
 *     `running` state and the §8.5 lock-timeout reclamation path
 *     will pick it back up after the next cutoff.
 *
 *   - {@see SyncJobRepository::claimNext()} failures (driver
 *     errors during the claim scan or conditional UPDATE) bubble
 *     out unchanged for the same reason: the worker has no
 *     meaningful recovery beyond surfacing the failure to its
 *     caller.
 *
 * Scope boundaries:
 *
 *   - The worker only dispatches `sync_product` jobs because the
 *     MVP only registers that one handler. A future slice can
 *     introduce a job-type → handler map and route on
 *     {@see SyncJobRecord::jobType()}; until then, jobs of any
 *     other type are skipped (their `claimed_at` lock will
 *     time-out and the row will be reclaimed for a future worker
 *     that knows how to dispatch them, matching ADR-0021 §8.2).
 *
 *   - This class does NOT log or write diagnostics. Those
 *     responsibilities belong to the surrounding cron service
 *     provider so the worker stays framework-free and trivially
 *     unit-testable; callers that want aggregate observability
 *     inspect the returned {@see SyncWorkerTickResult}.
 *
 *   - The worker DOES emit one §10.1 `sync_job_terminal` audit
 *     row per handled job when an {@see AuditEventWriter} is
 *     injected at construction time (Phase 2 KI-006 follow-up).
 *     Per-job audit emission stays inside the worker — and not
 *     on the surrounding cron provider — because the
 *     `(SyncJobRecord, SyncJobOutcome)` pair only co-exists
 *     inside the per-iteration loop; bubbling both back through
 *     {@see SyncWorkerTickResult} would either expand the result
 *     VO with handler-internal fields (jobId, productId, error
 *     code) or force the cron provider to re-decode the payload.
 *     The writer is optional: when null, the worker stays silent
 *     on the audit channel and the cron provider's existing
 *     "inspect the tick result" path is unchanged. Audit emit
 *     failures are swallowed inside {@see self::emitJobAudit()}
 *     per the {@see AuditEventWriter} contract so a failing
 *     audit-side persistence call cannot derail the surrounding
 *     tick.
 *
 * @package Affilun\Application\Sync
 */

declare(strict_types=1);

namespace Affilun\Application\Sync;

use Affilun\Application\Audit\AuditEventDraft;
use Affilun\Application\Audit\AuditEventWriter;
use Affilun\Domain\Clock;
use Affilun\Infrastructure\Persistence\SyncJobRecord;
use Affilun\Infrastructure\Persistence\SyncJobRepository;
use DateTimeImmutable;
use DateTimeZone;
use InvalidArgumentException;
use RuntimeException;

/**
 * Single-tick sync worker driver.
 *
 * See the file-level docblock for the per-tick claim → handle loop,
 * the lock-timeout cutoff arithmetic, and the failure semantics.
 */
final class SyncWorker {

	/**
	 * Default per-tick batch size per ADR-0021 §8.1 / spec §11.4
	 * "10〜50 件程度" guidance, biased toward the small end so a
	 * misconfigured WP-Cron tick cannot starve other request paths.
	 */
	public const DEFAULT_BATCH_SIZE = 5;

	/**
	 * Default spacing between handled sync jobs.
	 *
	 * Amazon Creators API documents rate limiting for getItems; the
	 * public docs are account-gated, but the observed 429 response
	 * points to the official api-rates page. Keep a safety margin over
	 * the common 1 request/second limit.
	 */
	public const DEFAULT_JOB_SPACING_MILLISECONDS = 1250;

	/**
	 * Default lock-timeout window per spec §8.5 / ADR-0021 §8.2
	 * "一定時間以上 `running` のままならタイムアウト回収対象". Five
	 * minutes is long enough to absorb a slow provider call without
	 * thrashing reclamation, and short enough that a crashed worker
	 * does not strand a job for the operator's next business day.
	 */
	public const DEFAULT_LOCK_TIMEOUT_SECONDS = 300;

	/**
	 * Job-type discriminator the worker dispatches in this slice.
	 *
	 * Mirrors {@see SyncJobRecord::createNew()}'s default type value
	 * so a future job-type → handler map can drop in without
	 * renaming this constant.
	 */
	public const JOB_TYPE_SYNC_PRODUCT = 'sync_product';

	/**
	 * Persistence port the worker drains.
	 */
	private SyncJobRepository $jobs;

	/**
	 * Application-layer handler the worker dispatches each claimed
	 * job to.
	 */
	private SyncProductJobHandler $handler;

	/**
	 * Clock port that produces the per-tick "now" instant.
	 */
	private Clock $clock;

	/**
	 * Stable per-process lock-holder identifier stamped onto
	 * `affilun_sync_jobs.locked_by` for every claim this worker makes.
	 */
	private string $workerId;

	/**
	 * Maximum jobs to drain per {@see self::tick()} invocation.
	 */
	private int $batchSize;

	/**
	 * Lock-timeout window in seconds; see
	 * {@see self::DEFAULT_LOCK_TIMEOUT_SECONDS}.
	 */
	private int $lockTimeoutSeconds;

	/**
	 * Minimum delay between handled jobs in a single worker tick.
	 */
	private int $jobSpacingMilliseconds;

	/**
	 * Wall-clock sleeper used to enforce {@see self::$jobSpacingMilliseconds}.
	 *
	 * @var callable(int): void
	 */
	private $sleeper;

	/**
	 * Optional §10.1 audit-event writer.
	 *
	 * When non-null, every handled job (success / retry_scheduled
	 * / failed_terminal) emits one `sync_job_terminal` audit row
	 * through this writer carrying the job id, the resolved
	 * `product_id` from the payload, and the outcome's error code
	 * / error message slots. When null (the default) the worker
	 * stays silent — pure-unit tests and bootstrap paths that
	 * pre-date the audit-writer port keep working without
	 * modification.
	 *
	 * Per the {@see AuditEventWriter} contract, audit emit
	 * failures are tolerated: a {@see RuntimeException} from
	 * `record()` is swallowed inside {@see self::emitJobAudit()}
	 * so losing one audit row never derails the surrounding tick
	 * (the job's terminal-state row in `affilun_sync_jobs` is the
	 * authoritative record either way).
	 */
	private ?AuditEventWriter $auditWriter;

	/**
	 * Construct the worker with explicit collaborators.
	 *
	 * @param SyncJobRepository     $jobs               Persistence port.
	 * @param SyncProductJobHandler $handler            Job handler.
	 * @param Clock                 $clock              Clock port.
	 * @param string                $workerId           Stable
	 *                                                  per-process
	 *                                                  lock-holder
	 *                                                  identifier
	 *                                                  (1–64 bytes).
	 * @param int                   $batchSize          Per-tick
	 *                                                  drain cap
	 *                                                  (must be > 0).
	 * @param int                   $lockTimeoutSeconds Reclamation
	 *                                                  threshold
	 *                                                  in seconds
	 *                                                  (must be > 0).
	 * @param AuditEventWriter|null $auditWriter        Optional
	 *                                                  §10.1 audit
	 *                                                  writer; when
	 *                                                  null, the
	 *                                                  worker does
	 *                                                  not emit
	 *                                                  audit rows.
	 * @param int                   $jobSpacingMilliseconds Minimum
	 *                                                  delay between
	 *                                                  handled sync
	 *                                                  jobs.
	 * @param callable|null         $sleeper            Optional
	 *                                                  test seam that
	 *                                                  receives the
	 *                                                  delay in
	 *                                                  microseconds.
	 *
	 * @throws InvalidArgumentException When `$workerId` is empty or
	 *                                  longer than 64 bytes, when
	 *                                  `$batchSize <= 0`, or when
	 *                                  `$lockTimeoutSeconds <= 0`.
	 */
	public function __construct(
		SyncJobRepository $jobs,
		SyncProductJobHandler $handler,
		Clock $clock,
		string $workerId,
		int $batchSize = self::DEFAULT_BATCH_SIZE,
		int $lockTimeoutSeconds = self::DEFAULT_LOCK_TIMEOUT_SECONDS,
		?AuditEventWriter $auditWriter = null,
		int $jobSpacingMilliseconds = self::DEFAULT_JOB_SPACING_MILLISECONDS,
		?callable $sleeper = null
	) {
		$length = strlen( $workerId );
		if ( $length === 0 || $length > 64 ) {
			$message = sprintf(
				'SyncWorker requires workerId of 1-64 bytes, got %d.',
				$length
			);
			// phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Developer-facing message, never rendered to HTML.
			throw new InvalidArgumentException( $message );
		}

		if ( $batchSize <= 0 ) {
			$message = sprintf(
				'SyncWorker requires batchSize > 0, got %d.',
				$batchSize
			);
			// phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Developer-facing message, never rendered to HTML.
			throw new InvalidArgumentException( $message );
		}

		if ( $lockTimeoutSeconds <= 0 ) {
			$message = sprintf(
				'SyncWorker requires lockTimeoutSeconds > 0, got %d.',
				$lockTimeoutSeconds
			);
			// phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Developer-facing message, never rendered to HTML.
			throw new InvalidArgumentException( $message );
		}

		if ( $jobSpacingMilliseconds < 0 ) {
			$message = sprintf(
				'SyncWorker requires jobSpacingMilliseconds >= 0, got %d.',
				$jobSpacingMilliseconds
			);
			// phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Developer-facing message, never rendered to HTML.
			throw new InvalidArgumentException( $message );
		}

		$this->jobs                   = $jobs;
		$this->handler                = $handler;
		$this->clock                  = $clock;
		$this->workerId               = $workerId;
		$this->batchSize              = $batchSize;
		$this->lockTimeoutSeconds     = $lockTimeoutSeconds;
		$this->auditWriter            = $auditWriter;
		$this->jobSpacingMilliseconds = $jobSpacingMilliseconds;
		$this->sleeper                = $sleeper ?? static function ( int $microseconds ): void {
			if ( $microseconds > 0 ) {
				usleep( $microseconds );
			}
		};
	}

	/**
	 * Stable per-process lock-holder identifier.
	 */
	public function workerId(): string {
		return $this->workerId;
	}

	/**
	 * Per-tick drain cap.
	 */
	public function batchSize(): int {
		return $this->batchSize;
	}

	/**
	 * Lock-timeout reclamation window in seconds.
	 */
	public function lockTimeoutSeconds(): int {
		return $this->lockTimeoutSeconds;
	}

	/**
	 * Minimum delay between handled jobs in a single tick.
	 */
	public function jobSpacingMilliseconds(): int {
		return $this->jobSpacingMilliseconds;
	}

	/**
	 * Drain up to {@see self::$batchSize} jobs and return the
	 * resulting tick summary.
	 *
	 * Stamps a single UTC "now" instant at the top of the tick, then
	 * loops `claim → handle → accumulate` until either the queue is
	 * idle (`claimNext()` returns null) or the batch budget is
	 * exhausted. Jobs of an unrecognized type are released back to
	 * the queue's lock-timeout reclamation path by skipping them
	 * without invoking the handler — they will time-out alongside
	 * the §8.5 stale-lock cohort.
	 */
	public function tick(): SyncWorkerTickResult {
		$now    = $this->clock->now()->setTimezone( new DateTimeZone( 'UTC' ) );
		$cutoff = $now->modify( '-' . $this->lockTimeoutSeconds . ' seconds' );

		$claimed                = 0;
		$skipped                = 0;
		$outcomes               = array();
		$handledSyncProductJobs = 0;

		for ( $i = 0; $i < $this->batchSize; $i++ ) {
			$record = $this->jobs->claimNext( $this->workerId, $now, $cutoff );
			if ( $record === null ) {
				break;
			}

			++$claimed;

			if ( $record->jobType() !== self::JOB_TYPE_SYNC_PRODUCT ) {
				++$skipped;
				continue;
			}

			if ( $handledSyncProductJobs > 0 ) {
				$this->sleepBetweenJobs();
			}

			$outcome    = $this->handler->handle( $record );
			$outcomes[] = $outcome;
			++$handledSyncProductJobs;

			$this->emitJobAudit( $record, $outcome );
		}

		return new SyncWorkerTickResult( $claimed, $skipped, $outcomes, $now );
	}

	/**
	 * Enforce the configured per-tick job spacing.
	 */
	private function sleepBetweenJobs(): void {
		if ( $this->jobSpacingMilliseconds <= 0 ) {
			return;
		}

		$microseconds = $this->jobSpacingMilliseconds * 1000;
		( $this->sleeper )( $microseconds );
	}

	/**
	 * Emit a `sync_job_terminal` audit row describing the
	 * just-handled job's outcome.
	 *
	 * No-op when no audit writer was injected at construction
	 * time. Failure to persist the row is swallowed per the
	 * {@see AuditEventWriter} "tolerate audit emit failure"
	 * contract — losing one audit row never derails the
	 * surrounding tick (the job's terminal-state row in
	 * `affilun_sync_jobs` is the authoritative record either way).
	 *
	 * Outcome → §10.2 result token mapping:
	 *
	 *   - {@see SyncJobOutcome::OUTCOME_SUCCESS}
	 *     → {@see AuditEventDraft::RESULT_SUCCESS} with
	 *     `event_type='sync.job.completed'` and `action='complete'`.
	 *
	 *   - {@see SyncJobOutcome::OUTCOME_RETRY_SCHEDULED}
	 *     → {@see AuditEventDraft::RESULT_FAILED} with
	 *     `event_type='sync.job.retry_scheduled'` and
	 *     `action='retry'`. Retry is modeled as a non-terminal
	 *     failure so the audit trail surfaces every transient
	 *     hiccup; the `nextAvailableAt` instant is carried in the
	 *     context payload so an operator can correlate the row
	 *     with the upcoming retry tick.
	 *
	 *   - {@see SyncJobOutcome::OUTCOME_FAILED_TERMINAL}
	 *     → {@see AuditEventDraft::RESULT_FAILED} with
	 *     `event_type='sync.job.failed'` and `action='fail'`.
	 *
	 * The draft uses {@see AuditEventDraft::forJobAction()} so
	 * the §12.4 envelope projection's `job_id` lift surfaces the
	 * sync-job id on the read side. The resolved `product_id`
	 * (extracted from the job payload) lands as the `target_id`
	 * with `target_type='product'` so the per-target audit
	 * drilldown groups every retry / success / failure for the
	 * same product together.
	 *
	 * @param SyncJobRecord  $record  Job record the worker just
	 *                                handled (carries the job id
	 *                                and payload).
	 * @param SyncJobOutcome $outcome Handler outcome to encode.
	 */
	private function emitJobAudit( SyncJobRecord $record, SyncJobOutcome $outcome ): void {
		if ( $this->auditWriter === null ) {
			return;
		}

		$jobId     = $record->id();
		$productId = self::extractProductId( $record->payloadJson() );

		try {
			if ( $outcome->isSuccess() ) {
				$eventType = 'sync.job.completed';
				$action    = 'complete';
				$result    = AuditEventDraft::RESULT_SUCCESS;
				$summary   = sprintf(
					'Sync job %s completed (attempts=%d).',
					$jobId === null ? '(unpersisted)' : (string) $jobId,
					$outcome->attempts()
				);
				$context   = array(
					'attempts' => $outcome->attempts(),
				);
			} elseif ( $outcome->isRetry() ) {
				$eventType = 'sync.job.retry_scheduled';
				$action    = 'retry';
				$result    = AuditEventDraft::RESULT_FAILED;
				$summary   = sprintf(
					'Sync job %s scheduled for retry (attempts=%d).',
					$jobId === null ? '(unpersisted)' : (string) $jobId,
					$outcome->attempts()
				);
				$context   = array(
					'attempts'          => $outcome->attempts(),
					'error_code'        => $outcome->errorCode(),
					'next_available_at' => $outcome->nextAvailableAt() === null
						? null
						: $outcome->nextAvailableAt()->format( DATE_ATOM ),
				);
			} else {
				// Terminal failure.
				$eventType = 'sync.job.failed';
				$action    = 'fail';
				$result    = AuditEventDraft::RESULT_FAILED;
				$summary   = sprintf(
					'Sync job %s failed terminally (attempts=%d).',
					$jobId === null ? '(unpersisted)' : (string) $jobId,
					$outcome->attempts()
				);
				$context   = array(
					'attempts'      => $outcome->attempts(),
					'error_code'    => $outcome->errorCode(),
					'error_message' => $outcome->errorMessage(),
				);
			}

			$draft = AuditEventDraft::forJobAction(
				$eventType,
				'sync_job_terminal',
				$action,
				$result,
				$summary,
				$jobId === null ? '(unpersisted)' : (string) $jobId,
				$productId === null ? null : 'product',
				$productId === null ? null : (string) $productId,
				array(),
				$context
			);

			$this->auditWriter->record( $draft );
		} catch ( RuntimeException $unused ) {
			// Per the AuditEventWriter contract, audit emit failure
			// must not derail the surrounding worker tick — the
			// job's terminal-state row in affilun_sync_jobs is the
			// authoritative record either way.
			unset( $unused );
		}
	}

	/**
	 * Extract the `product_id` field from the supplied JSON
	 * payload string, returning null when the field is missing,
	 * malformed, or non-positive.
	 *
	 * Mirrors the equivalent helper on
	 * {@see SyncProductJobHandler} so the audit row can carry the
	 * same `target_id` value the handler resolved from the
	 * payload, without coupling the worker to the handler's
	 * private surface.
	 *
	 * @param string $payloadJson Job payload JSON string.
	 */
	private static function extractProductId( string $payloadJson ): ?int {
		$decoded = json_decode( $payloadJson, true );
		if ( ! is_array( $decoded ) ) {
			return null;
		}
		if ( ! array_key_exists( 'product_id', $decoded ) ) {
			return null;
		}
		$raw = $decoded['product_id'];
		if ( ! is_int( $raw ) && ! ( is_string( $raw ) && ctype_digit( $raw ) ) ) {
			return null;
		}
		$id = (int) $raw;
		return $id > 0 ? $id : null;
	}

	/**
	 * Convenience accessor exposed for tests and diagnostics so
	 * downstream code does not have to re-derive the cutoff
	 * arithmetic when validating §8.5 reclamation behavior.
	 *
	 * @param DateTimeImmutable $now Reference instant.
	 */
	public function lockTimeoutCutoffFor( DateTimeImmutable $now ): DateTimeImmutable {
		return $now
			->setTimezone( new DateTimeZone( 'UTC' ) )
			->modify( '-' . $this->lockTimeoutSeconds . ' seconds' );
	}
}