407 lines
15 KiB
PHP
407 lines
15 KiB
PHP
<?php
|
|
|
|
namespace Modules\EcommerceFinance\Repositories;
|
|
|
|
use Illuminate\Database\ConnectionInterface;
|
|
use Illuminate\Database\DatabaseManager;
|
|
use Illuminate\Database\Query\Builder;
|
|
use Illuminate\Pagination\LengthAwarePaginator;
|
|
use Illuminate\Support\Carbon;
|
|
use Illuminate\Support\Facades\DB;
|
|
use Illuminate\Support\Facades\Schema;
|
|
use RuntimeException;
|
|
|
|
class ProductPriceMirrorRepository
|
|
{
|
|
public const DATASET_KEY = 'supplier-payables.product-prices';
|
|
|
|
public const EDITABLE_FIELDS = [
|
|
'barcode',
|
|
'spu',
|
|
'product_name',
|
|
'category',
|
|
'production_status',
|
|
'minimum_price',
|
|
'price_limit',
|
|
'purchase_price',
|
|
'shipping_fee',
|
|
'jingxi_limit_price',
|
|
];
|
|
|
|
private readonly ConnectionInterface $connection;
|
|
|
|
public function __construct(DatabaseManager $database)
|
|
{
|
|
$this->connection = $database->connection(
|
|
(string) config('upstream_mirrors.connection', 'business_mirror')
|
|
);
|
|
}
|
|
|
|
public function ready(): bool
|
|
{
|
|
$schema = Schema::connection($this->connection->getName());
|
|
|
|
return $schema->hasTable('business_mirror_sync_states')
|
|
&& $schema->hasTable('product_price_mirror_rows');
|
|
}
|
|
|
|
public function initialized(): bool
|
|
{
|
|
$state = $this->state();
|
|
|
|
return is_array($state) && ($state['initial_completed_at'] ?? null) !== null;
|
|
}
|
|
|
|
public function maintenanceReady(): bool
|
|
{
|
|
$schema = Schema::connection($this->connection->getName());
|
|
|
|
return $schema->hasTable('product_price_overrides')
|
|
&& $schema->hasTable('product_price_override_events');
|
|
}
|
|
|
|
/** @return array<string,mixed>|null */
|
|
public function state(): ?array
|
|
{
|
|
$row = $this->connection->table('business_mirror_sync_states')
|
|
->where('dataset_key', self::DATASET_KEY)
|
|
->first();
|
|
|
|
return $row === null ? null : (array) $row;
|
|
}
|
|
|
|
public function paginate(array $filters, int $page, int $limit, ?int $companyId = null): LengthAwarePaginator
|
|
{
|
|
$page = max($page, 1);
|
|
$limit = min(max($limit, 1), 100);
|
|
[$query, $withOverrides] = $this->effectiveQuery($companyId ?? (int) (company_id() ?: 1));
|
|
$this->applyFilters($query, $filters, $withOverrides);
|
|
$this->applySelect($query, $withOverrides);
|
|
|
|
return $this->applyOrder($query, $withOverrides)
|
|
->paginate($limit, ['*'], 'page', $page);
|
|
}
|
|
|
|
public function exportRows(array $filters, ?int $companyId = null): iterable
|
|
{
|
|
[$query, $withOverrides] = $this->effectiveQuery($companyId ?? (int) (company_id() ?: 1));
|
|
$this->applyFilters($query, $filters, $withOverrides);
|
|
$this->applySelect($query, $withOverrides);
|
|
|
|
return $this->applyOrder($query, $withOverrides)->cursor();
|
|
}
|
|
|
|
/**
|
|
* @param array<string,array<string,mixed>> $records keyed by upstream key
|
|
* @return array{created:int,updated:int}
|
|
*/
|
|
public function applyOverrides(
|
|
array $records,
|
|
int $companyId,
|
|
?int $userId,
|
|
string $source,
|
|
): array {
|
|
if (! $this->maintenanceReady()) {
|
|
throw new RuntimeException('product_price_maintenance_schema_missing');
|
|
}
|
|
|
|
if ($records === []) {
|
|
return ['created' => 0, 'updated' => 0];
|
|
}
|
|
|
|
$keys = array_values(array_unique(array_keys($records)));
|
|
|
|
return $this->connection->transaction(function () use (
|
|
$records,
|
|
$keys,
|
|
$companyId,
|
|
$userId,
|
|
$source,
|
|
): array {
|
|
[$effective, $withOverrides] = $this->effectiveQuery($companyId);
|
|
$this->applySelect($effective, $withOverrides);
|
|
$current = $effective->whereIn('r.upstream_key', $keys)->get()->keyBy('upstream_key');
|
|
|
|
if ($current->count() !== count($keys)) {
|
|
$missing = collect($keys)->first(fn (string $key): bool => ! $current->has($key));
|
|
throw new RuntimeException('product_price_record_not_found:' . $missing);
|
|
}
|
|
|
|
$existing = $this->connection->table('product_price_overrides')
|
|
->where('company_id', $companyId)
|
|
->whereIn('upstream_key', $keys)
|
|
->pluck('upstream_key')
|
|
->flip();
|
|
$created = 0;
|
|
$updated = 0;
|
|
$now = now();
|
|
|
|
foreach ($records as $upstreamKey => $changes) {
|
|
$row = (array) $current->get($upstreamKey);
|
|
$before = $this->editableValues($row);
|
|
$provided = array_intersect_key($changes, array_flip(self::EDITABLE_FIELDS));
|
|
$after = array_replace($before, $provided);
|
|
|
|
if (trim((string) ($after['barcode'] ?? '')) === '') {
|
|
throw new RuntimeException('product_price_barcode_required:' . $upstreamKey);
|
|
}
|
|
|
|
$identity = ['company_id' => $companyId, 'upstream_key' => $upstreamKey];
|
|
$values = $after + [
|
|
'source' => $source,
|
|
'updated_by' => $userId,
|
|
'updated_at' => $now,
|
|
];
|
|
|
|
if ($existing->has($upstreamKey)) {
|
|
$this->connection->table('product_price_overrides')->where($identity)->update($values);
|
|
$updated++;
|
|
} else {
|
|
$this->connection->table('product_price_overrides')->insert($identity + $values + [
|
|
'created_by' => $userId,
|
|
'created_at' => $now,
|
|
]);
|
|
$created++;
|
|
}
|
|
|
|
$this->connection->table('product_price_override_events')->insert([
|
|
'company_id' => $companyId,
|
|
'upstream_key' => $upstreamKey,
|
|
'source' => $source,
|
|
'before_values' => json_encode($before, JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION),
|
|
'after_values' => json_encode($after, JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION),
|
|
'user_id' => $userId,
|
|
'created_at' => $now,
|
|
]);
|
|
}
|
|
|
|
return compact('created', 'updated');
|
|
});
|
|
}
|
|
|
|
public function startRun(string $mode, Carbon $since, Carbon $until, bool $resume): array
|
|
{
|
|
$now = now();
|
|
$values = [
|
|
'source_key' => (string) config('upstream_mirrors.product_prices.source_key', 'operations_replica'),
|
|
'source_object' => (string) config('upstream_mirrors.product_prices.source_object', 'rpt.Bulk_Price_Limit'),
|
|
'status' => 'running',
|
|
'run_mode' => $mode,
|
|
'last_attempted_at' => $now,
|
|
'last_error_code' => null,
|
|
'updated_at' => $now,
|
|
];
|
|
|
|
if (! $resume) {
|
|
$values += [
|
|
'window_started_at' => $since,
|
|
'window_ends_at' => $until,
|
|
'cursor_completed_at' => null,
|
|
'cursor_business_id' => null,
|
|
'cursor_product_code' => null,
|
|
'last_upserted_rows' => 0,
|
|
'created_at' => $now,
|
|
];
|
|
}
|
|
|
|
$query = $this->connection->table('business_mirror_sync_states')
|
|
->where('dataset_key', self::DATASET_KEY);
|
|
|
|
if ($query->exists()) {
|
|
$query->update($values);
|
|
} else {
|
|
$this->connection->table('business_mirror_sync_states')->insert(
|
|
['dataset_key' => self::DATASET_KEY, 'created_at' => $now] + $values,
|
|
);
|
|
}
|
|
|
|
return $this->state() ?? [];
|
|
}
|
|
|
|
public function storeBatch(array $rows, array $cursor, int $cumulativeRows): void
|
|
{
|
|
$now = now();
|
|
$mapped = array_map(function (array $row) use ($now): array {
|
|
$businessId = (string) ($row['business_id'] ?? '');
|
|
$barcode = (string) ($row['barcode'] ?? '');
|
|
$cursorCompletedAt = (string) ($row['cursor_completed_at'] ?? $row['completed_at'] ?? '');
|
|
$values = [
|
|
'business_id' => $businessId,
|
|
'barcode' => $barcode,
|
|
'spu' => $row['spu'] ?? null,
|
|
'product_name' => $row['product_name'] ?? null,
|
|
'category' => $row['category'] ?? null,
|
|
'production_status' => $row['production_status'] ?? null,
|
|
'minimum_price' => $row['minimum_price'] ?? null,
|
|
'price_limit' => $row['price_limit'] ?? null,
|
|
'purchase_price' => $row['purchase_price'] ?? null,
|
|
'shipping_fee' => $row['shipping_fee'] ?? null,
|
|
'jingxi_limit_price' => $row['jingxi_limit_price'] ?? null,
|
|
'approval_status' => $row['approval_status'] ?? null,
|
|
'approval_result' => $row['approval_result'] ?? null,
|
|
'workflow_title' => $row['workflow_title'] ?? null,
|
|
'completed_at' => $row['completed_at'],
|
|
];
|
|
|
|
return $values + [
|
|
'upstream_key' => hash('sha256', $cursorCompletedAt . '|' . $businessId . '|' . $barcode),
|
|
'row_hash' => hash('sha256', json_encode($values, JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION)),
|
|
'synced_at' => $now,
|
|
'created_at' => $now,
|
|
'updated_at' => $now,
|
|
];
|
|
}, $rows);
|
|
|
|
$this->connection->transaction(function () use ($mapped, $cursor, $cumulativeRows, $now): void {
|
|
if ($mapped !== []) {
|
|
$this->connection->table('product_price_mirror_rows')->upsert(
|
|
$mapped,
|
|
['upstream_key'],
|
|
[
|
|
'business_id', 'barcode', 'spu', 'product_name', 'category',
|
|
'production_status', 'minimum_price', 'price_limit', 'purchase_price',
|
|
'shipping_fee', 'jingxi_limit_price', 'approval_status', 'approval_result',
|
|
'workflow_title', 'completed_at', 'row_hash', 'synced_at', 'updated_at',
|
|
],
|
|
);
|
|
}
|
|
|
|
$this->connection->table('business_mirror_sync_states')
|
|
->where('dataset_key', self::DATASET_KEY)
|
|
->update([
|
|
'cursor_completed_at' => $cursor['completed_at'],
|
|
'cursor_business_id' => $cursor['business_id'],
|
|
'cursor_product_code' => $cursor['barcode'],
|
|
'last_upserted_rows' => $cumulativeRows,
|
|
'updated_at' => $now,
|
|
]);
|
|
});
|
|
}
|
|
|
|
public function complete(string $mode, Carbon $windowEnd, int $rows): void
|
|
{
|
|
$now = now();
|
|
$values = [
|
|
'status' => 'succeeded',
|
|
'last_succeeded_at' => $windowEnd,
|
|
'last_upserted_rows' => $rows,
|
|
'cursor_completed_at' => null,
|
|
'cursor_business_id' => null,
|
|
'cursor_product_code' => null,
|
|
'last_error_code' => null,
|
|
'updated_at' => $now,
|
|
];
|
|
|
|
if ($mode === 'full') {
|
|
$values['initial_completed_at'] = $now;
|
|
}
|
|
|
|
$this->connection->table('business_mirror_sync_states')
|
|
->where('dataset_key', self::DATASET_KEY)
|
|
->update($values);
|
|
}
|
|
|
|
public function fail(string $errorCode): void
|
|
{
|
|
$this->connection->table('business_mirror_sync_states')
|
|
->where('dataset_key', self::DATASET_KEY)
|
|
->update([
|
|
'status' => 'failed',
|
|
'last_error_code' => mb_substr($errorCode, 0, 80),
|
|
'updated_at' => now(),
|
|
]);
|
|
}
|
|
|
|
private function contains(string $value): string
|
|
{
|
|
return '%' . str_replace(['!', '%', '_'], ['!!', '!%', '!_'], trim($value)) . '%';
|
|
}
|
|
|
|
/** @return array{0:Builder,1:bool} */
|
|
private function effectiveQuery(int $companyId): array
|
|
{
|
|
$query = $this->connection->table('product_price_mirror_rows as r');
|
|
$withOverrides = $this->maintenanceReady();
|
|
|
|
if ($withOverrides) {
|
|
$query->leftJoin('product_price_overrides as o', function ($join) use ($companyId): void {
|
|
$join->on('o.upstream_key', '=', 'r.upstream_key')
|
|
->where('o.company_id', '=', $companyId);
|
|
});
|
|
}
|
|
|
|
return [$query, $withOverrides];
|
|
}
|
|
|
|
private function applyFilters(Builder $query, array $filters, bool $withOverrides): void
|
|
{
|
|
if (($filters['barcode'] ?? '') !== '') {
|
|
$query->whereRaw(
|
|
$this->effectiveColumn('barcode', $withOverrides) . " LIKE ? ESCAPE '!'",
|
|
[$this->contains((string) $filters['barcode'])],
|
|
);
|
|
}
|
|
|
|
if (($filters['completed_from'] ?? '') !== '') {
|
|
$query->where('r.completed_at', '>=', (string) $filters['completed_from'] . ' 00:00:00');
|
|
}
|
|
|
|
if (($filters['completed_to'] ?? '') !== '') {
|
|
$query->where('r.completed_at', '<', Carbon::parse((string) $filters['completed_to'])->addDay()->startOfDay());
|
|
}
|
|
}
|
|
|
|
private function applySelect(Builder $query, bool $withOverrides): void
|
|
{
|
|
$query->select([
|
|
'r.id',
|
|
'r.upstream_key',
|
|
'r.business_id',
|
|
'r.approval_status',
|
|
'r.approval_result',
|
|
'r.workflow_title',
|
|
'r.completed_at',
|
|
]);
|
|
|
|
foreach (self::EDITABLE_FIELDS as $field) {
|
|
$query->addSelect(DB::raw($this->effectiveColumn($field, $withOverrides) . ' AS ' . $field));
|
|
}
|
|
|
|
if ($withOverrides) {
|
|
$query->addSelect([
|
|
DB::raw('CASE WHEN o.id IS NULL THEN 0 ELSE 1 END AS is_overridden'),
|
|
'o.updated_at as override_updated_at',
|
|
]);
|
|
} else {
|
|
$query->addSelect([
|
|
DB::raw('0 AS is_overridden'),
|
|
DB::raw('NULL AS override_updated_at'),
|
|
]);
|
|
}
|
|
}
|
|
|
|
private function applyOrder(Builder $query, bool $withOverrides): Builder
|
|
{
|
|
return $query
|
|
->orderByDesc('r.completed_at')
|
|
->orderByDesc('r.business_id')
|
|
->orderByRaw($this->effectiveColumn('barcode', $withOverrides) . ' ASC');
|
|
}
|
|
|
|
private function effectiveColumn(string $field, bool $withOverrides): string
|
|
{
|
|
return $withOverrides
|
|
? "CASE WHEN o.id IS NULL THEN r.{$field} ELSE o.{$field} END"
|
|
: "r.{$field}";
|
|
}
|
|
|
|
/** @return array<string,mixed> */
|
|
private function editableValues(array $row): array
|
|
{
|
|
return array_combine(
|
|
self::EDITABLE_FIELDS,
|
|
array_map(static fn (string $field): mixed => $row[$field] ?? null, self::EDITABLE_FIELDS),
|
|
);
|
|
}
|
|
}
|