get(); if ($services->isEmpty()) { $this->info('No subscription services configured. Exiting.'); return; } foreach ($services as $service) { $this->info("Starting sync for Service: {$service->name}"); // 1. Define High Water Mark (Fallback to an old date for the very first run) $lastSync = $service->last_synced_at ?? '2000-01-01 00:00:00'; // 2. Capture the exact time we are starting THIS sync $syncStartTime = now(); // 3. Inject Dynamic Connection try { Config::set('database.connections.service_dynamic', [ 'driver' => 'mysql', 'host' => $service->db_host, 'port' => $service->db_port, 'database' => $service->db_name, 'username' => $service->db_user, // 'password' => decrypt($service->db_password), 'password' => Crypt::decryptString($service->db_password), 'charset' => 'utf8mb4', 'collation' => 'utf8mb4_unicode_ci', ]); DB::purge('service_dynamic'); $dynamicDb = DB::connection('service_dynamic'); // Quick ping to verify credentials before running massive queries $dynamicDb->getPdo(); } catch (Exception $e) { $this->error("Connection failed for {$service->name}: " . $e->getMessage()); continue; // Skip to the next service on failure } // 4. Extract & Transform incrementally based on schema type try { if ($service->schema_type === 'type_a') { $this->syncTypeA($dynamicDb, $service->id, $lastSync); } elseif ($service->schema_type === 'type_b') { $this->syncTypeB($dynamicDb, $service->id, $lastSync); } else { $this->warn("Unknown schema type '{$service->schema_type}' for {$service->name}. Skipping."); continue; } // 5. Update the High Water Mark in the Hub database on success DB::table('subscription_services') ->where('id', $service->id) ->update(['last_synced_at' => $syncStartTime]); $this->info("Successfully synced {$service->name} up to {$syncStartTime}"); } catch (Exception $e) { // If it fails, the last_synced_at is not updated. // The next run will safely pick up from the previous successful mark. $this->error("Failed syncing {$service->name}: " . $e->getMessage()); } } $this->newLine(); $this->info('ETL Synchronization complete.'); } /** * Mapper for Schema Type A (msisdn, is_active, updated_at) */ private function syncTypeA($db, $serviceId, $lastSync) { $db->table('subscriptions') //->where('updated_at', '>=', $lastSync) ->orderBy('id') ->chunk(1000, function ($subscriptions) use ($serviceId) { $normalizedData = []; foreach ($subscriptions as $subscription) { $normalizedData[] = [ 'service_id' => $serviceId, 'phone_number' => $subscription->msisdn, 'content_name' => $subscription->content ?? 'Standard Bundle', 'status' => ($subscription->status == 'ON') ? 'active' : 'inactive', 'join_date' => $subscription->created_at, 'updated_at' => now(), 'created_at' => now(), ]; } // Bulk Upsert: Insert if new, Update if phone_number+service_id exists DB::table('unified_subscribers')->upsert( $normalizedData, ['service_id', 'phone_number', 'content_name'], ['status', 'updated_at'] ); }); } /** * Mapper for Schema Type B (cell_number, sub_state, updated_on) */ private function syncTypeB($db, $serviceId, $lastSync) { $db->table('subscribers') ->where('updated_on', '>=', $lastSync) ->orderBy('sub_id') ->chunk(1000, function ($subs) use ($serviceId) { $normalizedData = []; foreach ($subs as $sub) { $normalizedData[] = [ 'service_id' => $serviceId, 'phone_number' => $sub->cell_number, 'status' => strtolower($sub->sub_state), // standardizes to 'active' or 'inactive' 'join_date' => $sub->date_joined, 'updated_at' => now(), 'created_at' => now(), ]; } DB::table('unified_subscribers')->upsert( $normalizedData, ['service_id', 'phone_number'], ['status', 'updated_at'] ); }); } }