fix: atomic FTP writes, c2s file loss on crash
This commit is contained in:
SajjadMahmoody
parent
0e16ea6052
commit
ddad4da95c
|
|
@ -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(
|
||||
|
|
|
|||
Loading…
Reference in New Issue