Subscription-Content-Management/app/Console/Commands/SyncSubscriptionsServices.php

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Config;
use Exception;
use Illuminate\Support\Facades\Crypt;

class SyncSubscriptionsServices extends Command
{
    // The command name to run in the terminal
    protected $signature = 'subscriptions:sync';
    protected $description = 'Incremental ETL process to normalize and warehouse subscription data';

    public function handle()
    {
        $services = DB::table('subscription_services')->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']
                );
            });
    }
}