CaiWuXiTong/modules/EcommerceFinance/Repositories/DingWorklogMirrorRepository.php

<?php

namespace Modules\EcommerceFinance\Repositories;

use Carbon\CarbonInterface;
use Illuminate\Database\ConnectionInterface;
use Illuminate\Database\DatabaseManager;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\Schema;
use Throwable;

class DingWorklogMirrorRepository
{
    public const TABLE = 'ding_worklog_reports';

    public const STATE_KEY = 'dingtalk.worklog.finance_daily_reports';

    private readonly ConnectionInterface $connection;

    public function __construct(DatabaseManager $database)
    {
        $this->connection = $database->connection(
            (string) config('upstream_mirrors.connection', 'business_mirror')
        );
    }

    public function available(): bool
    {
        $schema = Schema::connection($this->connection->getName());

        return $schema->hasTable(self::TABLE)
            && $schema->hasTable('business_mirror_table_states');
    }

    /** @return array<string,mixed>|null */
    public function state(): ?array
    {
        if (! $this->available()) {
            return null;
        }

        $row = $this->connection->table('business_mirror_table_states')
            ->where('dataset_key', self::STATE_KEY)
            ->first();

        return $row === null ? null : (array) $row;
    }

    /** @param list<array<string,mixed>> $rows */
    public function upsertReports(array $rows): int
    {
        if ($rows === []) {
            return 0;
        }

        $this->connection->table(self::TABLE)->upsert(
            $rows,
            ['report_id'],
            [
                'template_name',
                'creator_userid',
                'creator_name',
                'submitted_at',
                'statistics_date',
                'remark',
                'raw_payload',
                'synced_at',
                'updated_at',
            ],
        );

        return count($rows);
    }

    public function recordSuccess(int $upserted, string $templateName, ?CarbonInterface $from, CarbonInterface $to): void
    {
        $now = now();
        $state = $this->state();
        $values = [
            'source_key' => 'dingtalk_worklog',
            'source_object' => $templateName !== '' ? $templateName : '钉钉日志',
            'target_table' => self::TABLE,
            'status' => 'succeeded',
            'run_mode' => 'incremental',
            'cursor' => json_encode([
                'from' => $from?->toDateTimeString(),
                'to' => Carbon::instance($to)->toDateTimeString(),
            ], JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR),
            'high_watermark' => json_encode(
                ['submitted_at' => Carbon::instance($to)->toDateTimeString()],
                JSON_THROW_ON_ERROR,
            ),
            'window_started_at' => $from ? Carbon::instance($from) : null,
            'window_ends_at' => Carbon::instance($to),
            'initial_completed_at' => $state['initial_completed_at'] ?? $now,
            'last_attempted_at' => $now,
            'last_succeeded_at' => $now,
            'last_upserted_rows' => $upserted,
            'mirror_row_count' => $this->connection->table(self::TABLE)->count(),
            'last_error_code' => null,
            'updated_at' => $now,
        ];

        $this->connection->table('business_mirror_table_states')->updateOrInsert(
            ['dataset_key' => self::STATE_KEY],
            $values + ['created_at' => $state['created_at'] ?? $now],
        );
    }

    public function recordFailure(string $errorCode): void
    {
        if (! $this->available()) {
            return;
        }

        $now = now();
        $state = $this->state();
        $values = [
            'source_key' => 'dingtalk_worklog',
            'source_object' => (string) config('dingtalk.worklog.template_name', '财务日报'),
            'target_table' => self::TABLE,
            'status' => 'failed',
            'run_mode' => 'incremental',
            'last_attempted_at' => $now,
            'last_error_code' => mb_substr($errorCode, 0, 120),
            'updated_at' => $now,
        ];

        $this->connection->table('business_mirror_table_states')->updateOrInsert(
            ['dataset_key' => self::STATE_KEY],
            $values + [
                'initial_completed_at' => $state['initial_completed_at'] ?? null,
                'mirror_row_count' => $this->connection->table(self::TABLE)->count(),
                'created_at' => $state['created_at'] ?? $now,
            ],
        );
    }

    /**
     * @return list<array{
     *   statistics_date:string,
     *   submission_count:int,
     *   submitter_count:int,
     *   first_submitted_at:?string,
     *   last_submitted_at:?string
     * }>
     */
    public function dailySummary(string $from, string $to): array
    {
        return $this->connection->table(self::TABLE)
            ->whereBetween('statistics_date', [$from, $to])
            ->groupBy('statistics_date')
            ->orderByDesc('statistics_date')
            ->selectRaw(
                'statistics_date, COUNT(*) AS submission_count, '
                . 'COUNT(DISTINCT creator_userid) AS submitter_count, '
                . 'MIN(submitted_at) AS first_submitted_at, MAX(submitted_at) AS last_submitted_at'
            )
            ->get()
            ->map(static fn (object $row): array => [
                'statistics_date' => (string) $row->statistics_date,
                'submission_count' => (int) $row->submission_count,
                'submitter_count' => (int) $row->submitter_count,
                'first_submitted_at' => $row->first_submitted_at ? (string) $row->first_submitted_at : null,
                'last_submitted_at' => $row->last_submitted_at ? (string) $row->last_submitted_at : null,
            ])
            ->values()
            ->all();
    }

    /** @return list<object> */
    public function submissionsForDate(string $date, int $limit = 200): array
    {
        return $this->connection->table(self::TABLE)
            ->where('statistics_date', $date)
            ->orderByDesc('submitted_at')
            ->limit(min(max($limit, 1), 500))
            ->get([
                'report_id',
                'template_name',
                'creator_userid',
                'creator_name',
                'submitted_at',
                'statistics_date',
            ])
            ->all();
    }

    public static function statisticsDate(CarbonInterface|int|string $submittedAt): Carbon
    {
        $submitted = self::submittedAt($submittedAt)->timezone(
            (string) config('app.timezone', 'Asia/Shanghai')
        );

        return $submitted->hour < 12
            ? $submitted->copy()->subDay()->startOfDay()
            : $submitted->copy()->startOfDay();
    }

    public static function submittedAt(CarbonInterface|int|string $value): Carbon
    {
        $timezone = (string) config('app.timezone', 'Asia/Shanghai');

        if ($value instanceof CarbonInterface) {
            return Carbon::instance($value)->timezone($timezone);
        }

        if (is_int($value) || (is_string($value) && ctype_digit($value))) {
            $timestamp = (int) $value;

            return Carbon::createFromTimestampMs(
                $timestamp >= 1000000000000 ? $timestamp : $timestamp * 1000,
                $timezone,
            );
        }

        return Carbon::parse((string) $value, $timezone)->timezone($timezone);
    }

    public static function reportId(array $row): string
    {
        $reportId = trim((string) ($row['report_id'] ?? $row['id'] ?? ''));
        if ($reportId !== '') {
            return mb_substr($reportId, 0, 64);
        }

        return hash('sha256', json_encode($row, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
    }

    public static function isPermissionError(?string $errorCode): bool
    {
        return $errorCode === 'dingtalk_worklog_permission_required';
    }

    public function permissionApplyUrl(): string
    {
        $clientId = trim((string) config('dingtalk.client_id', ''));
        if ($clientId === '') {
            return 'https://open-dev.dingtalk.com/appscope/apply';
        }

        return 'https://open-dev.dingtalk.com/appscope/apply?content='
            . rawurlencode($clientId . '#' . (string) config('dingtalk.worklog.permission_scope', 'qyapi_report_query'));
    }
}