<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\Redis;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Carbon\Carbon;

class ProcessRedisRealtimeEntriesCommand extends Command
{
    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'redis:process:realtime {--batch=500 : Number of events per DB insert batch}';

    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = 'Process realtime Redis events and move them into MySQL table `asterisk_events` then clear the Redis list.';

    public function __construct()
    {
        parent::__construct();
    }

    public function handle()
    {
        $stream = 'realtime:events';
        $group = 'realtime-workers';
        $consumer = 'processor-' . gethostname() . '-' . getmypid();
        $batchSize = (int)$this->option('batch') ?: 500;

        // Ensure consumer group exists (create if missing)
        try {
            try {
                Redis::xgroup('CREATE', $stream, $group, '0', 'MKSTREAM');
                $this->info("Created consumer group {$group} on stream {$stream}");
            } catch (\Exception $e) {
                // BUSYGROUP means it already exists — ignore
                if (strpos($e->getMessage(), 'BUSYGROUP') === false) {
                    // Other error — log and continue (group may already exist)
                    $this->info('xgroup create: ' . $e->getMessage());
                }
            }
        } catch (\Exception $e) {
            $this->error('Failed to ensure consumer group: ' . $e->getMessage());
            return 1;
        }

        $this->info('Processing stream ' . $stream . ' with group ' . $group . ' as consumer ' . $consumer);

        $processed = 0;
        $insertFailures = 0;

        try {
            // First process any pending entries (delivered but not ACKed)
            while (true) {
                $entries = Redis::xreadgroup($group, $consumer, [$stream => '0'], $batchSize, 0);
                if (empty($entries)) break;
                $ids = [];
                $rows = [];
                foreach ($entries as $sName => $items) {
                    foreach ($items as $id => $fields) {
                        $ids[] = $id;
                        // fields expected to contain key 'data'
                        $raw = null;
                        if (is_array($fields)) {
                            if (isset($fields['data'])) $raw = $fields['data'];
                            else {
                                // sometimes returned as list
                                $flat = is_array($fields) ? $fields : [];
                                if (count($flat) >= 2) {
                                    // attempt to extract value after 'data' key
                                    $raw = end($flat);
                                }
                            }
                        }
                        $d = null;
                        if ($raw) $d = json_decode($raw, true);
                        if (!is_array($d)) {
                            $rows[] = [
                                'event_type' => null,
                                'raw' => $raw ?? json_encode($fields),
                                'meta' => null,
                                'created_at' => Carbon::now(),
                            ];
                            continue;
                        }

                        $eventType = $d['type'] ?? $d['event'] ?? ($d['event_type'] ?? null);
                        $rawField = $d['raw'] ?? (isset($d['raw']) ? $d['raw'] : json_encode($d));
                        $meta = $d['meta'] ?? (isset($d['meta']) ? $d['meta'] : null);
                        $createdAt = isset($d['ts']) ? Carbon::parse($d['ts']) : Carbon::now();

                        $rows[] = [
                            'event_type' => $eventType,
                            'raw' => $rawField,
                            'meta' => $meta,
                            'created_at' => $createdAt,
                        ];
                    }
                }

                if (!empty($rows)) {
                    try {
                        DB::table('asterisk_events')->insert($rows);
                        $processed += count($rows);
                        $this->info('Inserted batch: ' . count($rows));
                    } catch (\Exception $e) {
                        $insertFailures += count($rows);
                        $this->error('DB insert failed for pending batch: ' . $e->getMessage());
                        return 1;
                    }
                }

                // ACK and optionally delete
                try {
                    if (!empty($ids)) {
                        // xack
                        Redis::xack($stream, $group, $ids);
                        // xdel to free space
                        Redis::xdel($stream, $ids);
                    }
                } catch (\Exception $e) {
                    Log::warning('Failed to xack/xdel pending ids: ' . $e->getMessage());
                }
            }

            // Now read new entries (non-blocking until none left)
            while (true) {
                $entries = Redis::xreadgroup($group, $consumer, [$stream => '>'], $batchSize, 0);
                if (empty($entries)) break;
                $ids = [];
                $rows = [];
                foreach ($entries as $sName => $items) {
                    foreach ($items as $id => $fields) {
                        $ids[] = $id;
                        $raw = null;
                        if (is_array($fields)) {
                            if (isset($fields['data'])) $raw = $fields['data'];
                            else {
                                $flat = is_array($fields) ? $fields : [];
                                if (count($flat) >= 2) $raw = end($flat);
                            }
                        }
                        $d = null;
                        if ($raw) $d = json_decode($raw, true);
                        if (!is_array($d)) {
                            $rows[] = [
                                'event_type' => null,
                                'raw' => $raw ?? json_encode($fields),
                                'meta' => null,
                                'created_at' => Carbon::now(),
                            ];
                            continue;
                        }

                        $eventType = $d['type'] ?? $d['event'] ?? ($d['event_type'] ?? null);
                        $rawField = $d['raw'] ?? (isset($d['raw']) ? $d['raw'] : json_encode($d));
                        $meta = $d['meta'] ?? (isset($d['meta']) ? $d['meta'] : null);
                        $createdAt = isset($d['ts']) ? Carbon::parse($d['ts']) : Carbon::now();

                        $rows[] = [
                            'event_type' => $eventType,
                            'raw' => $rawField,
                            'meta' => $meta,
                            'created_at' => $createdAt,
                        ];
                    }
                }

                if (!empty($rows)) {
                    try {
                        DB::table('asterisk_events')->insert($rows);
                        $processed += count($rows);
                        $this->info('Inserted batch: ' . count($rows));
                    } catch (\Exception $e) {
                        $insertFailures += count($rows);
                        $this->error('DB insert failed for new batch: ' . $e->getMessage());
                        return 1;
                    }
                }

                // ACK and delete
                try {
                    if (!empty($ids)) {
                        Redis::xack($stream, $group, $ids);
                        Redis::xdel($stream, $ids);
                    }
                } catch (\Exception $e) {
                    Log::warning('Failed to xack/xdel new ids: ' . $e->getMessage());
                }
            }

            $this->info('Processing complete. Processed: ' . $processed . ' events. Insert failures: ' . $insertFailures);

            // update simple stats
            try {
                Redis::set('realtime:stats:last_processed_at', Carbon::now()->toDateTimeString());
                Redis::incrby('realtime:stats:total_processed', $processed);
            } catch (\Exception $e) {
                Log::warning('Failed to update realtime stats in Redis: ' . $e->getMessage());
            }

            return 0;

        } catch (\Exception $e) {
            $msg = $e->getMessage();
            // Fallback: if Redis client does not support Streams commands, revert to list-based processing
            if (stripos($msg, 'not a registered Redis command') !== false || stripos($msg, 'XREADGROUP') !== false || stripos($msg, 'XGROUP') !== false) {
                $this->info('Redis does not support Streams commands in this client; falling back to list-processing. (' . $msg . ')');
                return $this->processListFallback('realtime:events', $batchSize);
            }
            $this->error('Unexpected processing error: ' . $msg);
            return 1;
        }
    }

    /**
     * Fallback list-based processing (used when Redis Streams commands aren't available).
     */
    private function processListFallback($key, $batchSize)
    {
        try {
            $len = Redis::llen($key);
        } catch (\Exception $e) {
            $this->error('Failed to query Redis: ' . $e->getMessage());
            return 1;
        }

        if (empty($len)) {
            $this->info('No events to process (list fallback).');
            return 0;
        }

        $this->info("Found {$len} event(s) in {$key} (list fallback)");

        $tempKey = $key . ':processing:' . time() . ':' . bin2hex(random_bytes(4));

        try {
            if (!Redis::exists($key)) {
                $this->info('No events present at rename time.');
                return 0;
            }
            Redis::rename($key, $tempKey);
        } catch (\Exception $e) {
            $this->error('Failed to rename Redis key for processing: ' . $e->getMessage());
            return 1;
        }

        $this->info('Renamed key to ' . $tempKey . ' for processing (list fallback)');

        try {
            $total = Redis::llen($tempKey);
            $processed = 0;
            $insertFailures = 0;

            for ($offset = 0; $offset < $total; $offset += $batchSize) {
                $stop = min($offset + $batchSize - 1, $total - 1);
                $items = Redis::lrange($tempKey, $offset, $stop);
                if (empty($items)) continue;

                $rows = [];
                foreach ($items as $raw) {
                    $d = json_decode($raw, true);
                    if (!is_array($d)) {
                        $rows[] = [
                            'event_type' => null,
                            'raw' => $raw,
                            'meta' => null,
                            'created_at' => Carbon::now(),
                        ];
                        continue;
                    }

                    $eventType = $d['type'] ?? $d['event'] ?? ($d['event_type'] ?? null);
                    $rawField = $d['raw'] ?? (isset($d['raw']) ? $d['raw'] : json_encode($d));
                    $meta = $d['meta'] ?? (isset($d['meta']) ? $d['meta'] : null);
                    $createdAt = isset($d['ts']) ? Carbon::parse($d['ts']) : Carbon::now();

                    $rows[] = [
                        'event_type' => $eventType,
                        'raw' => $rawField,
                        'meta' => $meta,
                        'created_at' => $createdAt,
                    ];
                }

                try {
                    DB::table('asterisk_events')->insert($rows);
                    $processed += count($rows);
                    $this->info('Inserted batch: ' . count($rows));
                } catch (\Exception $e) {
                    $insertFailures += count($rows);
                    $this->error('DB insert failed for batch starting at ' . $offset . ': ' . $e->getMessage());
                    // try to restore
                    try {
                        if (Redis::exists($tempKey)) {
                            if (!Redis::exists($key)) {
                                Redis::rename($tempKey, $key);
                            } else {
                                $remaining = Redis::lrange($tempKey, 0, -1);
                                foreach ($remaining as $r) {
                                    Redis::rpush($key, $r);
                                }
                                Redis::del($tempKey);
                            }
                        }
                    } catch (\Exception $re) {
                        $this->error('Failed to restore Redis key after DB error: ' . $re->getMessage());
                    }
                    return 1;
                }
            }

            Redis::del($tempKey);
            $this->info('Processing complete (list fallback). Processed: ' . $processed . ' events. Insert failures: ' . $insertFailures);

            try {
                Redis::set('realtime:stats:last_processed_at', Carbon::now()->toDateTimeString());
                Redis::incrby('realtime:stats:total_processed', $processed);
            } catch (\Exception $e) {
                Log::warning('Failed to update realtime stats in Redis: ' . $e->getMessage());
            }

            return 0;

        } catch (\Exception $e) {
            try {
                if (Redis::exists($tempKey)) {
                    if (!Redis::exists($key)) {
                        Redis::rename($tempKey, $key);
                    } else {
                        $remaining = Redis::lrange($tempKey, 0, -1);
                        foreach ($remaining as $r) Redis::rpush($key, $r);
                        Redis::del($tempKey);
                    }
                }
            } catch (\Exception $re) {
                $this->error('Failed to restore Redis list after unexpected error: ' . $re->getMessage());
            }
            $this->error('Unexpected processing error (list fallback): ' . $e->getMessage());
            return 1;
        }
    }
}