From ddad4da95c4ef966081a05f6d7df3d29c2607a9d Mon Sep 17 00:00:00 2001 From: SajjadMahmoody <117562560+SajjadMahmoody@users.noreply.github.com> Date: Tue, 2 Jun 2026 01:18:25 +0330 Subject: [PATCH] fix: atomic FTP writes, c2s file loss on crash --- app/Services/SyncService.php | 163 ++++++++++++++++++++++++----------- 1 file changed, 114 insertions(+), 49 deletions(-) diff --git a/app/Services/SyncService.php b/app/Services/SyncService.php index 2eaf775..3862167 100644 --- a/app/Services/SyncService.php +++ b/app/Services/SyncService.php @@ -83,6 +83,11 @@ public function sync(): array $report = []; $baseUrl = "http://{$this->peerIp}:{$this->peerPort}"; + $healed = $this->healMissingFiles(); + if ($healed > 0) { + $report[] = "{$healed} فایل گمشده شناسایی شد و در صف دریافت قرار گرفت."; + } + $peerCursor = \App\Models\OsurgInitial::val('sync_peer_cursor') ?: null; $ourLastSync = \App\Models\OsurgInitial::val('our_sync_cursor') ?: null; @@ -155,6 +160,7 @@ public function sync(): array $ourDatetimes = array_filter(array_column($ourChanges, 'datetime')); $maxOurDatetime = \Carbon\Carbon::parse(max($ourDatetimes))->format('Y-m-d H:i:s'); if ($ourLastSync === null || $maxOurDatetime > $ourLastSync) { + $this->preQueueOutgoingFiles($ourChanges); $this->saveOurCursor($maxOurDatetime); } } elseif ($hasErrors) { @@ -194,26 +200,9 @@ public function applyChanges(array $changes): array return $report; } - private function transferFilesViaFtp(array $changes, string $side, array &$report): void + protected function transferFilesViaFtp(array $changes, string $side, array &$report): void { - $fileFields = ['photos_before', 'photos_after', 'videos', 'audio_files']; - - $currentPaths = []; - foreach ($changes as $change) { - $data = $change['changed_data'] ?? []; - foreach ($fileFields as $field) { - if (empty($data[$field])) { - continue; - } - $files = is_array($data[$field]) ? $data[$field] : json_decode($data[$field], true) ?? []; - foreach ($files as $filePath) { - if ($filePath) { - $currentPaths[] = ltrim(str_replace('\\', '/', $filePath), '/'); - } - } - } - } - + $currentPaths = $this->extractFilePaths($changes); $pendingPaths = $this->getPendingFiles($side); $allPaths = array_values(array_unique(array_merge($pendingPaths, $currentPaths))); @@ -254,7 +243,7 @@ private function transferFilesViaFtp(array $changes, string $side, array &$repor continue; } - if (ftp_put($ftp, $remoteFile, $localFile, FTP_BINARY)) { + if ($this->ftpPut($ftp, $remoteFile, $localFile)) { $report[] = "فایل [{$filePath}] ارسال شد (FTP) ✓"; Log::info('FTP file uploaded', ['file' => $filePath, 'size' => filesize($localFile)]); } else { @@ -286,9 +275,9 @@ private function transferFilesViaFtp(array $changes, string $side, array &$repor } } - if (ftp_get($ftp, $localFile, $remoteFile, FTP_BINARY)) { + if ($this->ftpGet($ftp, $localFile, $remoteFile)) { $report[] = "فایل [{$filePath}] دریافت شد (FTP) ✓"; - Log::info('FTP file downloaded', ['file' => $filePath, 'size' => filesize($localFile)]); + Log::info('FTP file downloaded', ['file' => $filePath, 'size' => file_exists($localFile) ? filesize($localFile) : 0]); } else { $errorDetail = error_get_last(); $errorMsg = "فایل [{$filePath}] دریافت نشد (FTP)"; @@ -310,11 +299,42 @@ private function transferFilesViaFtp(array $changes, string $side, array &$repor $report[] = count($failedPaths) . ' فایل در صف retry دفعه بعد قرار گرفت.'; } - ftp_close($ftp); + $this->ftpClose($ftp); Log::info('FTP connection closed'); } - private function connectFtp(): mixed + protected function ftpGet(mixed $ftp, string $localFile, string $remoteFile): bool + { + $tmpFile = $localFile . '.part'; + + if (file_exists($tmpFile)) { + @unlink($tmpFile); + } + + $ok = @ftp_get($ftp, $tmpFile, $remoteFile, FTP_BINARY); + + if ($ok && file_exists($tmpFile)) { + return rename($tmpFile, $localFile); + } + + if (file_exists($tmpFile)) { + @unlink($tmpFile); + } + + return false; + } + + protected function ftpPut(mixed $ftp, string $remoteFile, string $localFile): bool + { + return ftp_put($ftp, $remoteFile, $localFile, FTP_BINARY); + } + + protected function ftpClose(mixed $ftp): void + { + ftp_close($ftp); + } + + protected function connectFtp(): mixed { if (! function_exists('ftp_connect')) { Log::error('FTP extension is not available'); @@ -351,7 +371,7 @@ private function connectFtp(): mixed return $ftp; } - private function createFtpDirectory($ftp, string $path): bool + protected function createFtpDirectory($ftp, string $path): bool { $parts = explode('/', trim($path, '/')); $currentPath = ''; @@ -384,11 +404,10 @@ private function createFtpDirectory($ftp, string $path): bool return true; } - private function pushFilesViaHttp(array $changes, string $baseUrl, array &$report): void + private function extractFilePaths(array $changes): array { $fileFields = ['photos_before', 'photos_after', 'videos', 'audio_files']; - - $currentPaths = []; + $paths = []; foreach ($changes as $change) { $data = $change['changed_data'] ?? []; foreach ($fileFields as $field) { @@ -398,11 +417,17 @@ private function pushFilesViaHttp(array $changes, string $baseUrl, array &$repor $files = is_array($data[$field]) ? $data[$field] : json_decode($data[$field], true) ?? []; foreach ($files as $filePath) { if ($filePath) { - $currentPaths[] = ltrim(str_replace('\\', '/', $filePath), '/'); + $paths[] = ltrim(str_replace('\\', '/', $filePath), '/'); } } } } + return $paths; + } + + private function pushFilesViaHttp(array $changes, string $baseUrl, array &$report): void + { + $currentPaths = $this->extractFilePaths($changes); $pendingPaths = $this->getPendingFiles('c2s'); $allPaths = array_values(array_unique(array_merge($pendingPaths, $currentPaths))); @@ -439,23 +464,7 @@ private function pushFilesViaHttp(array $changes, string $baseUrl, array &$repor private function syncFilesViaHttp(array $changes, string $baseUrl, array &$report): void { - $fileFields = ['photos_before', 'photos_after', 'videos', 'audio_files']; - - $currentPaths = []; - foreach ($changes as $change) { - $data = $change['changed_data'] ?? []; - foreach ($fileFields as $field) { - if (empty($data[$field])) { - continue; - } - $files = is_array($data[$field]) ? $data[$field] : json_decode($data[$field], true) ?? []; - foreach ($files as $filePath) { - if ($filePath) { - $currentPaths[] = ltrim(str_replace('\\', '/', $filePath), '/'); - } - } - } - } + $currentPaths = $this->extractFilePaths($changes); $pendingPaths = $this->getPendingFiles('s2c'); $allPaths = array_values(array_unique(array_merge($pendingPaths, $currentPaths))); @@ -520,17 +529,63 @@ private function applyChange(array $change, string $prefix, array &$report): voi } } - private function getPendingFiles(string $direction): array + + public function healMissingFiles(): int + { + $fileFields = ['photos_before', 'photos_after', 'videos', 'audio_files']; + $missing = []; + + $rows = DB::table('patients') + ->where(function ($q) use ($fileFields) { + foreach ($fileFields as $field) { + $q->orWhereNotNull($field); + } + }) + ->get($fileFields); + + foreach ($rows as $row) { + foreach ($fileFields as $field) { + if (empty($row->$field)) { + continue; + } + $files = json_decode($row->$field, true) ?? []; + foreach ($files as $filePath) { + if (! $filePath) { + continue; + } + $filePath = ltrim(str_replace('\\', '/', $filePath), '/'); + if (! Storage::disk('public')->exists($filePath)) { + $missing[] = $filePath; + } + } + } + } + + $missing = array_values(array_unique($missing)); + + if (! empty($missing)) { + $existing = $this->getPendingFiles('s2c'); + $merged = array_values(array_unique(array_merge($existing, $missing))); + $this->savePendingFiles('s2c', $merged); + Log::info('healMissingFiles: added to pending queue', ['count' => count($missing)]); + } + + return count($missing); + } + + protected function getPendingFiles(string $direction): array { $record = \App\Models\OsurgInitial::where('init_parameter', "sync_pending_{$direction}")->first(); if (! $record || ! $record->attachment) { return []; } - return json_decode($record->attachment, true) ?: []; + $decoded = json_decode($record->attachment, true); + + return is_array($decoded) && array_is_list($decoded) ? $decoded : []; } - private function savePendingFiles(string $direction, array $files): void + protected function savePendingFiles(string $direction, array $files): void { $files = array_values(array_unique(array_filter($files))); \App\Models\OsurgInitial::updateOrCreate( @@ -539,6 +594,16 @@ private function savePendingFiles(string $direction, array $files): void ); } + private function preQueueOutgoingFiles(array $changes): void + { + $paths = $this->extractFilePaths($changes); + if (empty($paths)) { + return; + } + $existing = $this->getPendingFiles('c2s'); + $this->savePendingFiles('c2s', array_merge($existing, $paths)); + } + private function saveOurCursor(string $cursor): void { \App\Models\OsurgInitial::updateOrCreate(