added subscriber profile

app/Console/Commands/SyncSubscriptionsServices.php

@@ -78,6 +78,8 @@ class SyncSubscriptionsServices extends Command
                 // The next run will safely pick up from the previous successful mark.
                 $this->error("Failed syncing {$service->name}: " . $e->getMessage());
             }
+            // Inside your service loop in SyncSubscriptionsServices.php
+            $this->syncBillingTransactions($dynamicDb, $service->id, $lastSync);
         }
 
         $this->newLine();
@@ -129,7 +131,8 @@ class SyncSubscriptionsServices extends Command
                     $normalizedData[] = [
                         'service_id'   => $serviceId,
                         'phone_number' => $sub->cell_number,
-                        'status'       => strtolower($sub->sub_state), // standardizes to 'active' or 'inactive'
+                        'content_name' => $sub->content ?? 'Standard Bundle', // Added to match composite key
+                        'status'       => strtolower($sub->sub_state), 
                         'join_date'    => $sub->date_joined,
                         'updated_at'   => now(),
                         'created_at'   => now(),
@@ -138,9 +141,40 @@ class SyncSubscriptionsServices extends Command
                 
                 DB::table('unified_subscribers')->upsert(
                     $normalizedData, 
-                    ['service_id', 'phone_number'], 
+                    ['service_id', 'phone_number', 'content_name'], // Updated to match table's unique index
                     ['status', 'updated_at']
                 );
             });
     }
+
+    private function syncBillingTransactions($db, $serviceId, $lastSync)
+    {
+        // Check if the remote database actually has the billing_transactions table
+        if (!$db->getSchemaBuilder()->hasTable('billing_transactions')) {
+            return;
+        }
+
+        $db->table('billing_transactions')
+            ->where('date', '>=', $lastSync) // Only pull transactions from the high-water mark forward
+            ->orderBy('id')
+            ->chunk(1000, function ($transactions) use ($serviceId) {
+                $batchData = [];
+                foreach ($transactions as $tx) {
+                    $batchData[] = [
+                        'service_id'       => $serviceId,
+                        'msisdn'           => $tx->msisdn,
+                        'content'          => $tx->content,
+                        'cost'             => $tx->cost,
+                        'transaction_date' => $tx->date,
+                        'created_at'       => now(),
+                        'updated_at'       => now(),
+                    ];
+                }
+                
+                // Insert log entries into the hub warehouse
+                if (!empty($batchData)) {
+                    DB::table('unified_billing_transactions')->insert($batchData);
+                }
+            });
+    }
 }