415 lines
24 KiB
PHP
415 lines
24 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
namespace app\common\service\prescriptionai;
|
|
|
|
use DomainException;
|
|
use think\facade\Db;
|
|
|
|
/** Database-backed outbox, immutable result storage and fenced work leases. */
|
|
final class PrescriptionAiStore
|
|
{
|
|
/** Cached per connection for rolling deployment before the additive migration arrives. */
|
|
public static function supportsProgress(): bool
|
|
{
|
|
static $columns = [];
|
|
$query = Db::name('prescription_ai_task');
|
|
$connection = $query->getConnection();
|
|
$key = spl_object_id($connection) . ':' . $query->getConfig('database');
|
|
if (!array_key_exists($key, $columns)) {
|
|
$columns[$key] = isset($query->getFields()['progress_json']);
|
|
}
|
|
return $columns[$key];
|
|
}
|
|
|
|
public static function enabled(): bool
|
|
{
|
|
return (bool) config('prescription_analysis.enabled', false);
|
|
}
|
|
|
|
public static function recordSaved(array $rx, int $actorId, ?array $info = null, array $options = []): ?int
|
|
{
|
|
if (!self::enabled() || !PrescriptionAiPolicy::isManual($rx)) {
|
|
return null;
|
|
}
|
|
$info = $info ?? PrescriptionAiAccess::actor($actorId);
|
|
if (!$info || !PrescriptionAiAccess::canGenerate($actorId, $info)) {
|
|
return null;
|
|
}
|
|
// Caller holds the prescription mutation transaction; failure must propagate.
|
|
return Db::transaction(static function () use ($rx, $actorId, $options): ?int {
|
|
$id = (int) $rx['id'];
|
|
$subject = Db::name('prescription_ai_subject')->where('prescription_id', $id)->lock(true)->find();
|
|
$hash = PrescriptionAiPolicy::fingerprint($rx);
|
|
if ($subject && hash_equals((string) $subject['clinical_hash'], $hash)
|
|
&& empty($options['restored'])) {
|
|
return (int) $subject['latest_batch_id'];
|
|
}
|
|
$revision = $subject ? (int) $subject['revision'] + 1 : 1;
|
|
$now = time();
|
|
if (!$subject) {
|
|
Db::name('prescription_ai_subject')->insert([
|
|
'prescription_id' => $id, 'revision' => $revision, 'clinical_hash' => $hash,
|
|
'first_batch_id' => 0, 'latest_batch_id' => 0, 'updated_at' => $now,
|
|
]);
|
|
} else {
|
|
self::invalidate($id, 'prescription_changed');
|
|
Db::name('prescription_ai_subject')->where('prescription_id', $id)->update([
|
|
'revision' => $revision, 'clinical_hash' => $hash, 'updated_at' => $now,
|
|
]);
|
|
}
|
|
$trigger = $subject ? 'clinical_change' : (string) ($options['trigger'] ?? 'first_manual');
|
|
if ($subject && !config('prescription_analysis.auto_refresh_prescription', true)) {
|
|
return null;
|
|
}
|
|
return self::insertBatch($rx, $revision, $actorId, $trigger,
|
|
hash('sha256', "saved:{$id}:{$revision}"), $options);
|
|
});
|
|
}
|
|
|
|
public static function enqueue(array $rx, int $actorId, string $trigger, string $eventKey, array $options = []): int
|
|
{
|
|
return Db::transaction(static function () use ($rx, $actorId, $trigger, $eventKey, $options): int {
|
|
$rx = Db::name('tcm_prescription')->where('id', (int) $rx['id'])->lock(true)->find();
|
|
if (!self::enabled() || !$rx || !PrescriptionAiPolicy::isManual($rx)) {
|
|
throw new DomainException('该处方不能安排分析');
|
|
}
|
|
$subject = Db::name('prescription_ai_subject')->where('prescription_id', (int) $rx['id'])->lock(true)->find();
|
|
if (!$subject || !hash_equals($subject['clinical_hash'], PrescriptionAiPolicy::fingerprint($rx))) {
|
|
$info = PrescriptionAiAccess::actor($actorId);
|
|
$id = self::recordSaved($rx, $actorId, $info, ['trigger' => $trigger] + $options);
|
|
if (!$id) {
|
|
throw new DomainException('无法为该处方安排分析');
|
|
}
|
|
return $id;
|
|
}
|
|
$existing = Db::name('prescription_ai_batch')->where('event_key', $eventKey)->value('id');
|
|
if ($existing) {
|
|
return (int) $existing;
|
|
}
|
|
$latest = Db::name('prescription_ai_batch')->where('id', (int) $subject['latest_batch_id'])->find();
|
|
if ($trigger === 'manual_refresh' && $latest && $latest['validity'] === 'current'
|
|
&& in_array($latest['status'], ['preparing', 'waiting_sources', 'queued', 'running'], true)) {
|
|
return (int) $latest['id'];
|
|
}
|
|
self::invalidate((int) $rx['id'], 'source_updated');
|
|
return self::insertBatch($rx, (int) $subject['revision'], $actorId, $trigger, $eventKey, $options);
|
|
});
|
|
}
|
|
|
|
private static function insertBatch(array $rx, int $revision, int $actorId, string $trigger, string $key, array $options): int
|
|
{
|
|
$now = time();
|
|
$diagnosisId = (int) ($rx['diagnosis_id'] ?? 0);
|
|
$patientId = $diagnosisId > 0 ? (int) Db::name('tcm_diagnosis')->where('id', $diagnosisId)
|
|
->whereNull('delete_time')->value('patient_id') : 0;
|
|
$bindingValid = $patientId > 0 && ((int) ($rx['patient_id'] ?? 0) === 0 || (int) $rx['patient_id'] === $patientId);
|
|
$aiAssisted = array_key_exists('ai_assisted', $options)
|
|
? (in_array($options['ai_assisted'], [true, 1, '1'], true) ? 'yes'
|
|
: (in_array($options['ai_assisted'], [false, 0, '0'], true) ? 'no' : 'unknown')) : 'unknown';
|
|
$id = (int) Db::name('prescription_ai_batch')->insertGetId([
|
|
'event_key' => $key, 'prescription_id' => (int) $rx['id'], 'prescription_revision' => $revision,
|
|
'clinical_hash' => PrescriptionAiPolicy::fingerprint($rx), 'diagnosis_id' => $diagnosisId,
|
|
'patient_id' => $patientId, 'doctor_id' => (int) ($rx['creator_id'] ?? 0), 'actor_id' => $actorId,
|
|
'trigger_type' => $trigger, 'reason' => mb_substr((string) ($options['reason'] ?? ''), 0, 500),
|
|
'ai_assisted' => $aiAssisted, 'status' => $bindingValid ? 'preparing' : 'blocked',
|
|
'validity' => 'current', 'comparison_type' => 'latest_context', 'baseline_eligible' => 0,
|
|
'prescription_cipher' => (new PrescriptionAiCipher())->encrypt($rx, 'prescription'),
|
|
'decision_at' => $now, 'wait_until' => $now + (int) config('prescription_analysis.transcript_wait_seconds', 300),
|
|
'next_run_at' => $now + ($trigger === 'clinical_change' ? (int) config('prescription_analysis.debounce_seconds', 60) : 0),
|
|
'error_code' => $bindingValid ? '' : 'PATIENT_BINDING_REQUIRED', 'created_at' => $now, 'updated_at' => $now,
|
|
]);
|
|
$subject = Db::name('prescription_ai_subject')->where('prescription_id', (int) $rx['id'])->find();
|
|
Db::name('prescription_ai_subject')->where('prescription_id', (int) $rx['id'])->update([
|
|
'latest_batch_id' => $id, 'first_batch_id' => (int) ($subject['first_batch_id'] ?? 0) ?: $id, 'updated_at' => $now,
|
|
]);
|
|
return $id;
|
|
}
|
|
|
|
public static function invalidate(int $rxId, string $validity): void
|
|
{
|
|
if (!self::enabled()) {
|
|
return;
|
|
}
|
|
$ids = Db::name('prescription_ai_batch')->where('prescription_id', $rxId)->where('validity', 'current')->column('id');
|
|
if ($ids === []) {
|
|
return;
|
|
}
|
|
$now = time();
|
|
Db::name('prescription_ai_batch')->whereIn('id', $ids)->update([
|
|
'validity' => $validity, 'lock_token' => '', 'lock_until' => 0, 'updated_at' => $now,
|
|
]);
|
|
Db::name('prescription_ai_batch')->whereIn('id', $ids)
|
|
->whereIn('status', ['preparing', 'waiting_sources', 'retry_wait', 'queued', 'running'])->update(['status' => 'cancelled']);
|
|
Db::name('prescription_ai_task')->whereIn('batch_id', $ids)->whereIn('status', PrescriptionAiPolicy::ACTIVE_TASKS)
|
|
->update(['status' => 'cancelled', 'lock_token' => '', 'lock_until' => 0, 'updated_at' => $now]);
|
|
$taskIds = Db::name('prescription_ai_task')->whereIn('batch_id', $ids)->column('id');
|
|
if ($taskIds !== []) {
|
|
Db::name('prescription_ai_attempt')->whereIn('task_id', $taskIds)->where('status', 'running')
|
|
->update(['status' => 'cancelled', 'finished_at' => $now]);
|
|
}
|
|
}
|
|
|
|
public static function claimBatch(): ?array
|
|
{
|
|
$now = time();
|
|
$ids = Db::name('prescription_ai_batch')->where('validity', 'current')
|
|
->whereIn('status', ['preparing', 'waiting_sources', 'retry_wait'])->where('next_run_at', '<=', $now)
|
|
->where('lock_until', '<=', $now)->order('id')->limit(20)->column('id');
|
|
foreach ($ids as $id) {
|
|
$row = Db::transaction(static function () use ($id, $now): ?array {
|
|
$row = Db::name('prescription_ai_batch')->where('id', $id)->lock(true)->find();
|
|
if (!$row || $row['validity'] !== 'current' || (int) $row['lock_until'] > $now
|
|
|| !in_array($row['status'], ['preparing', 'waiting_sources', 'retry_wait'], true)
|
|
|| (int) $row['next_run_at'] > $now) {
|
|
return null;
|
|
}
|
|
$row['lock_token'] = bin2hex(random_bytes(16));
|
|
$row['lock_until'] = $now + (int) config('prescription_analysis.lease_seconds', 600);
|
|
Db::name('prescription_ai_batch')->where('id', $id)->update([
|
|
'lock_token' => $row['lock_token'], 'lock_until' => $row['lock_until'], 'updated_at' => $now,
|
|
]);
|
|
return $row;
|
|
});
|
|
if ($row) {
|
|
return $row;
|
|
}
|
|
}
|
|
return null;
|
|
}
|
|
|
|
public static function finishPreparation(array $batch, array $context): bool
|
|
{
|
|
return Db::transaction(static function () use ($batch, $context): bool {
|
|
$current = Db::name('prescription_ai_batch')->where('id', $batch['id'])->lock(true)->find();
|
|
if (!self::owns($current, $batch)) {
|
|
return false;
|
|
}
|
|
$now = time();
|
|
if (!empty($context['wait_for_transcript']) && $now < (int) $current['wait_until']) {
|
|
self::releasePreparation($batch, ['status' => 'waiting_sources', 'next_run_at' => $now + 15]);
|
|
return true;
|
|
}
|
|
$exclusions = (array) ($context['baseline_exclusion_reasons'] ?? []);
|
|
if ($current['ai_assisted'] !== 'no') {
|
|
$exclusions[] = $current['ai_assisted'] === 'yes' ? 'ai_assisted' : 'assistance_unknown';
|
|
}
|
|
if ($current['trigger_type'] !== 'first_manual' && $current['trigger_type'] !== 'blank_to_manual') {
|
|
$exclusions[] = 'not_original_submission';
|
|
}
|
|
$eligible = !empty($context['baseline_eligible']) && $exclusions === [];
|
|
Db::name('prescription_ai_batch')->where('id', $batch['id'])->update([
|
|
'context_cipher' => (new PrescriptionAiCipher())->encrypt($context, 'context'),
|
|
'access_cipher' => (new PrescriptionAiCipher())->encrypt(
|
|
['source_access_manifest' => $context['source_access_manifest'] ?? []], 'access:' . $batch['id']),
|
|
'source_hash' => (string) ($context['source_hash'] ?? ''),
|
|
'source_diagnosis_ids_json' => PrescriptionAiPolicy::canonical($context['source_diagnosis_ids'] ?? []),
|
|
'source_summary_json' => PrescriptionAiPolicy::canonical($context['source_summary'] ?? []),
|
|
'missing_json' => PrescriptionAiPolicy::canonical($context['missing'] ?? []),
|
|
'comparison_type' => (string) ($context['comparison_type'] ?? 'latest_context'),
|
|
'baseline_eligible' => (int) $eligible,
|
|
'baseline_exclusions_json' => PrescriptionAiPolicy::canonical(array_values(array_unique($exclusions))),
|
|
'cutoff_at' => (int) ($context['cutoff_at'] ?? $now), 'status' => 'queued',
|
|
'coverage_status' => empty($context['missing']) ? 'pending' : 'partial',
|
|
'lock_token' => '', 'lock_until' => 0, 'error_code' => '', 'updated_at' => $now,
|
|
]);
|
|
foreach (PrescriptionAiPolicy::MODELS as $model) {
|
|
Db::name('prescription_ai_task')->insert([
|
|
'batch_id' => $batch['id'], 'model_key' => $model, 'status' => 'queued',
|
|
'next_run_at' => $now, 'updated_at' => $now,
|
|
]);
|
|
}
|
|
return true;
|
|
});
|
|
}
|
|
|
|
public static function releasePreparation(array $batch, array $values): bool
|
|
{
|
|
return (bool) Db::name('prescription_ai_batch')->where('id', $batch['id'])
|
|
->where('lock_token', $batch['lock_token'])->where('validity', 'current')->update($values + [
|
|
'lock_token' => '', 'lock_until' => 0, 'updated_at' => time(),
|
|
]);
|
|
}
|
|
|
|
public static function claimTask(string $model): ?array
|
|
{
|
|
if (!in_array($model, PrescriptionAiPolicy::MODELS, true)) {
|
|
throw new DomainException('未知模型');
|
|
}
|
|
$now = time();
|
|
$ids = Db::name('prescription_ai_task')->where('model_key', $model)->whereIn('status', PrescriptionAiPolicy::ACTIVE_TASKS)
|
|
->where('next_run_at', '<=', $now)->where('lock_until', '<=', $now)->order('id')->limit(30)->column('id');
|
|
foreach ($ids as $id) {
|
|
$result = Db::transaction(static function () use ($id, $model, $now): ?array {
|
|
// One locked daily model counter serializes concurrency and budget reservations.
|
|
$key = date('Y-m-d', $now) . ':' . $model;
|
|
Db::name('prescription_ai_limit')->extra('IGNORE')->insert(['limit_key' => $key, 'used_count' => 0, 'updated_at' => $now]);
|
|
$budget = Db::name('prescription_ai_limit')->where('limit_key', $key)->lock(true)->find();
|
|
$batchId = Db::name('prescription_ai_task')->where('id', $id)->value('batch_id');
|
|
// All mutations lock batch before task, matching invalidation and completion.
|
|
$batch = $batchId ? Db::name('prescription_ai_batch')->where('id', $batchId)->lock(true)->find() : null;
|
|
$row = Db::name('prescription_ai_task')->where('id', $id)->lock(true)->find();
|
|
if (!$row || !in_array($row['status'], PrescriptionAiPolicy::ACTIVE_TASKS, true)
|
|
|| (int) $row['lock_until'] > $now || (int) $row['next_run_at'] > $now) {
|
|
return null;
|
|
}
|
|
if (!$batch || $batch['validity'] !== 'current') {
|
|
Db::name('prescription_ai_task')->where('id', $id)->update(['status' => 'cancelled', 'updated_at' => $now]);
|
|
return null;
|
|
}
|
|
$limit = max(1, (int) config('prescription_analysis.max_attempts', 3));
|
|
if ($row['status'] === 'running') {
|
|
Db::name('prescription_ai_attempt')->where('task_id', $id)->where('status', 'running')
|
|
->update(['status' => 'expired', 'error_code' => 'LEASE_EXPIRED', 'finished_at' => $now]);
|
|
}
|
|
if ((int) $row['attempts'] >= $limit) {
|
|
Db::name('prescription_ai_task')->where('id', $id)->update([
|
|
'status' => 'failed', 'error_code' => 'LEASE_EXPIRED', 'finished_at' => $now, 'updated_at' => $now,
|
|
]);
|
|
self::refreshBatch((int) $row['batch_id']);
|
|
return null;
|
|
}
|
|
if ((int) $budget['used_count'] >= max(1, (int) config('prescription_analysis.daily_model_tasks', 200))) {
|
|
Db::name('prescription_ai_task')->where('id', $id)->update([
|
|
'status' => 'retry_wait', 'error_code' => 'BUDGET_PAUSED',
|
|
'next_run_at' => strtotime('tomorrow', $now), 'updated_at' => $now,
|
|
]);
|
|
return null;
|
|
}
|
|
$running = Db::name('prescription_ai_task')->where('model_key', $model)->where('status', 'running')
|
|
->where('lock_until', '>', $now)->count();
|
|
if ($running >= max(1, (int) config('prescription_analysis.max_parallel_per_model', 1))) {
|
|
return null;
|
|
}
|
|
$row['attempts'] = (int) $row['attempts'] + 1;
|
|
$row['total_attempts'] = (int) $row['total_attempts'] + 1;
|
|
$row['status'] = 'running';
|
|
$row['lock_token'] = bin2hex(random_bytes(16));
|
|
$row['lock_until'] = $now + (int) config('prescription_analysis.lease_seconds', 600);
|
|
$row['started_at'] = $now;
|
|
$row['finished_at'] = 0;
|
|
$public = self::supportsProgress() ? ['progress_json' => PrescriptionAiPolicy::canonical(
|
|
PrescriptionAiProgress::advance([], 'preparing', 'running', null, null, $now))] : [];
|
|
Db::name('prescription_ai_task')->where('id', $id)->update(array_intersect_key($row, array_flip([
|
|
'attempts', 'total_attempts', 'status', 'lock_token', 'lock_until', 'started_at', 'finished_at',
|
|
])) + $public + ['error_code' => '', 'updated_at' => $now]);
|
|
Db::name('prescription_ai_attempt')->insert([
|
|
'task_id' => $id, 'attempt_no' => $row['total_attempts'], 'status' => 'running', 'started_at' => $now,
|
|
]);
|
|
Db::name('prescription_ai_limit')->where('limit_key', $key)->inc('used_count')->update(['updated_at' => $now]);
|
|
Db::name('prescription_ai_batch')->where('id', $row['batch_id'])->update(['status' => 'running', 'updated_at' => $now]);
|
|
return $row;
|
|
});
|
|
if ($result) {
|
|
return $result;
|
|
}
|
|
}
|
|
return null;
|
|
}
|
|
|
|
public static function checkpoint(array $task, array $progress, bool $persistCache = true): bool
|
|
{
|
|
$now = time();
|
|
$values = ['lock_until' => $now + (int) config('prescription_analysis.lease_seconds', 600), 'updated_at' => $now];
|
|
if (self::supportsProgress()) {
|
|
$values['progress_json'] = PrescriptionAiPolicy::canonical(PrescriptionAiProgress::sanitize($progress['public'] ?? null));
|
|
}
|
|
// Stage notifications do not rewrite the growing encrypted model cache.
|
|
if ($persistCache) {
|
|
$values['progress_cipher'] = (new PrescriptionAiCipher())->encrypt($progress, 'progress:' . $task['id']);
|
|
}
|
|
$owned = static fn () => Db::name('prescription_ai_task')->where('id', $task['id'])->where('status', 'running')
|
|
->where('lock_token', $task['lock_token'])->where('lock_until', '>', time());
|
|
$changed = $owned()->update($values);
|
|
// Identical metadata twice in one second (also old-schema heartbeats) can be a no-op.
|
|
return $changed > 0 || $owned()->count() > 0;
|
|
}
|
|
|
|
public static function complete(array $task, array $output, array $comparison): bool
|
|
{
|
|
return Db::transaction(static function () use ($task, $output, $comparison): bool {
|
|
$rxId = Db::name('prescription_ai_batch')->where('id', $task['batch_id'])->value('prescription_id');
|
|
$rx = Db::name('tcm_prescription')->where('id', $rxId)->lock(true)->find();
|
|
$batch = Db::name('prescription_ai_batch')->where('id', $task['batch_id'])->lock(true)->find();
|
|
$row = Db::name('prescription_ai_task')->where('id', $task['id'])->lock(true)->find();
|
|
if (!self::owns($row, $task) || $row['status'] !== 'running') {
|
|
return false;
|
|
}
|
|
if (!$batch || $batch['validity'] !== 'current' || !$rx || !PrescriptionAiPolicy::isManual($rx)
|
|
|| !hash_equals($batch['clinical_hash'], PrescriptionAiPolicy::fingerprint($rx))) {
|
|
return false;
|
|
}
|
|
$coverage = $output['coverage'] ?? [];
|
|
$coverageStatus = is_array($coverage) ? (string) ($coverage['status'] ?? 'partial') : 'partial';
|
|
$coverageStatus = in_array($coverageStatus, ['complete', 'full'], true) ? 'complete' : 'partial';
|
|
$body = ['report' => $output['report'] ?? [], 'candidate' => $output['candidate'] ?? null,
|
|
'comparison' => $comparison, 'coverage' => $coverage, 'usage' => $output['usage'] ?? []];
|
|
$now = time();
|
|
$resultId = (int) Db::name('prescription_ai_result')->insertGetId([
|
|
'batch_id' => $task['batch_id'], 'model_key' => $task['model_key'],
|
|
'body_cipher' => (new PrescriptionAiCipher())->encrypt($body, 'result:' . $task['batch_id'] . ':' . $task['model_key']),
|
|
'score' => $comparison['score'] ?? null, 'herb_score' => $comparison['herb_score'] ?? null,
|
|
'comparison_status' => (string) ($comparison['status'] ?? 'not_comparable'),
|
|
'comparison_reason_code' => (string) ($comparison['reason_code'] ?? ''), 'coverage_status' => $coverageStatus,
|
|
'model_name' => mb_substr((string) ($output['model_name'] ?? ''), 0, 100),
|
|
'prompt_version' => mb_substr((string) ($output['prompt_version'] ?? ''), 0, 100),
|
|
'algorithm_version' => mb_substr((string) ($comparison['algorithm_version'] ?? ''), 0, 100), 'generated_at' => $now,
|
|
'dictionary_version' => mb_substr((string) ($comparison['dictionary_version'] ?? ''), 0, 100),
|
|
]);
|
|
$public = self::supportsProgress() ? ['progress_json' => PrescriptionAiPolicy::canonical(
|
|
PrescriptionAiProgress::advance([], 'completed', 'completed', null, null, $now))] : [];
|
|
Db::name('prescription_ai_task')->where('id', $task['id'])->update($public + [
|
|
'status' => 'success', 'result_id' => $resultId, 'finished_at' => $now, 'updated_at' => $now,
|
|
'lock_token' => '', 'lock_until' => 0, 'error_code' => '', 'progress_cipher' => null,
|
|
]);
|
|
Db::name('prescription_ai_attempt')->where('task_id', $task['id'])->where('attempt_no', $row['total_attempts'])
|
|
->update(['status' => 'success', 'finished_at' => $now]);
|
|
self::refreshBatch((int) $task['batch_id']);
|
|
return true;
|
|
});
|
|
}
|
|
|
|
public static function fail(array $task, string $code, bool $retryable): void
|
|
{
|
|
Db::transaction(static function () use ($task, $code, $retryable): void {
|
|
Db::name('prescription_ai_batch')->where('id', $task['batch_id'])->lock(true)->find();
|
|
$current = Db::name('prescription_ai_task')->where('id', $task['id'])->lock(true)->find();
|
|
if (!self::owns($current, $task) || $current['status'] !== 'running') {
|
|
return;
|
|
}
|
|
$now = time();
|
|
$code = preg_match('/^[A-Z0-9_]{1,64}$/D', $code) ? $code : 'INTERNAL_ERROR';
|
|
$next = PrescriptionAiPolicy::retryAt((int) $task['attempts'], $now, $retryable,
|
|
max(1, (int) config('prescription_analysis.max_attempts', 3)));
|
|
Db::name('prescription_ai_task')->where('id', $task['id'])->where('lock_token', $task['lock_token'])
|
|
->where('status', 'running')->update([
|
|
'status' => $next === null ? 'failed' : 'retry_wait', 'next_run_at' => $next ?? $now,
|
|
'error_code' => $code, 'lock_token' => '', 'lock_until' => 0, 'finished_at' => $now, 'updated_at' => $now,
|
|
]);
|
|
self::refreshBatch((int) $task['batch_id']);
|
|
Db::name('prescription_ai_attempt')->where('task_id', $task['id'])->where('attempt_no', $current['total_attempts'])
|
|
->update(['status' => 'failed', 'error_code' => $code, 'finished_at' => $now]);
|
|
});
|
|
}
|
|
|
|
public static function refreshBatch(int $batchId): void
|
|
{
|
|
$states = Db::name('prescription_ai_task')->where('batch_id', $batchId)->column('status');
|
|
$coverage = Db::name('prescription_ai_result')->where('batch_id', $batchId)->column('coverage_status');
|
|
Db::name('prescription_ai_batch')->where('id', $batchId)->where('validity', 'current')->update([
|
|
'status' => PrescriptionAiPolicy::aggregate($states),
|
|
'coverage_status' => count($coverage) === 2 && $coverage === ['complete', 'complete'] ? 'complete' : 'partial',
|
|
'updated_at' => time(),
|
|
]);
|
|
}
|
|
|
|
private static function owns(?array $current, array $claim): bool
|
|
{
|
|
return $current && !empty($claim['lock_token'])
|
|
&& hash_equals((string) $current['lock_token'], (string) $claim['lock_token'])
|
|
&& (int) $current['lock_until'] > time()
|
|
&& (!isset($current['validity']) || $current['validity'] === 'current');
|
|
}
|
|
}
|