Raghunathan Krishankumar

"""
REDB Ingestor - Core ingestion orchestration.

This module contains the Ingestor class which orchestrates the entire
sample processing pipeline: querying catalogs, downloading from S3,
dispatching to workers, and collecting results.

The actual implementation is split across focused modules:
- redb.queries: Database query and deduplication functions
- redb.s3_utils: S3/MinIO client and file operations
- redb.workers: File processing and worker functions
- redb.logging_utils: Logging setup and ImportResult enum

For backward compatibility, all public names from these modules are
re-exported here so that `from redb.ingestor import *` continues to work.
"""
import multiprocessing
from multiprocessing import Pool
from datetime import datetime
import os
import sys
import gc
import psutil
import time
import json
import tempfile
import warnings
from urllib3.exceptions import InsecureRequestWarning
from dotenv import load_dotenv

load_dotenv(override=True)

warnings.filterwarnings("ignore", category=InsecureRequestWarning, module="urllib3")
warnings.filterwarnings("ignore", category=UserWarning, module="elasticsearch")

# =============================================================================
# Re-exports for backward compatibility
# =============================================================================
# These imports ensure that `from redb.ingestor import X` and
# `@patch('redb.ingestor.X')` continue to work after the refactor.

from redb.logging_utils import (  # noqa: F401
    ImportResult,
    FileNameFormatter,
    setup_logger,
    logger_thread,
    setup_direct_logger,
)

from redb.queries import (  # noqa: F401
    get_supported_formats,
    get_db_catalog_connection,
    fetch_s3_objects_by_repository,
    fetch_s3_objects_by_date_range,
    fetch_analyzed_samples,
    is_in_db,
    is_in_code_db,
    is_in_db_bulk,
)

from redb.s3_utils import (  # noqa: F401
    get_minio_client,
    generate_s3_key_from_hash,
    download_s3_object,
    extract_fat_slices,
)

from redb.workers import (  # noqa: F401
    process_s3_file,
    process_file,
    _process_file_internal,
    process_zip_file,
    process_7zip_file,
    process_binary_file,
    worker,
    direct_s3_worker,
    is_binary_file,
    check_dotnet,
    check_high_swap,
    get_module_by_name,
    filter_selected_modules,
    _is_packed,
)

# Re-export settings for patches like @patch('redb.ingestor.settings')
from redb import settings  # noqa: F401

# Re-export hashlib and Magika for patches like @patch('redb.ingestor.hashlib')
import hashlib  # noqa: F401
try:
    from magika import Magika  # noqa: F401
except ImportError:
    pass

# Re-export py7zr for patches like @patch('redb.ingestor.py7zr')
try:
    import py7zr  # noqa: F401
except ImportError:
    pass


class Ingestor:
    def __init__(
        self,
        path=None,
        decompile=False,
        yara_scan=False,
        with_yara=False,
        repository="",
        index_prefix="",
        selected_modules=None,
        s3_mode=False,
        s3_notes=None,
        magika_filter=None,
        s3_solo=False,
        s3_solo_hash=None,
        s3_solo_key=None,
        dry_run=False,
        force=False,
        job_id=None,
        start_date=None,
        end_date=None,
        analyzed=False,
        decompile_modules=None,
        rerun=False,
    ):

        # Set multiprocessing start method as early as possible
        try:
            multiprocessing.set_start_method('spawn', force=True)
        except RuntimeError:
            current_method = multiprocessing.get_start_method()
            if current_method != 'spawn':
                print(f"[WARNING] Multiprocessing start method is {current_method}, not 'spawn'. This may cause issues.")

        self.path = path
        self.decompile = decompile
        self.yara_scan = yara_scan
        self.with_yara = with_yara
        self.repository = repository
        self.index_prefix = index_prefix
        self.selected_modules = selected_modules
        self.s3_mode = s3_mode
        self.s3_notes = s3_notes
        self.magika_filter = magika_filter
        self.s3_solo = s3_solo
        self.s3_solo_hash = s3_solo_hash
        self.s3_solo_key = s3_solo_key
        self.dry_run = dry_run
        # --rerun implies force at the worker level: the query already selects
        # only already-disassembled samples, so the per-file is_in_code_db
        # dedup check must be skipped or every sample gets skipped.
        self.force = force or rerun
        self.job_id = job_id
        self.start_date = start_date
        self.end_date = end_date
        self.analyzed = analyzed
        self.decompile_modules = decompile_modules or {"all"}
        self.rerun = rerun

        self.manager = multiprocessing.Manager()
        self.file_type_stats = self.manager.dict()
        self.total_results = self.manager.dict({result: 0 for result in ImportResult})

        self.today = datetime.today().strftime("%Y%m%dT%H%M%S")
        log_base_path = os.getenv("LOG_FILE_PATH", "/app/logs/")
        if not log_base_path.endswith("/"):
            log_base_path += "/"
        index_suffix = self.index_prefix.upper() if self.index_prefix else "DEFAULT"
        self.log_file = log_base_path + f"{self.today}-{self.repository}-{index_suffix}.txt"

        with open(self.log_file, "a") as f:
            f.write(f"CMD: {' '.join(sys.argv)}\n")
            f.write(f"=== Ingestor started at {datetime.now()} ===\n")


    def restart_worker_pool(self):
        """Restart the worker pool to help address memory issues"""
        if hasattr(self, 'pool') and self.pool:
            try:
                print("[INFO] Restarting worker pool to address memory fragmentation")
                self.pool.close()
                self.pool.join()
                self.pool = None
            except:
                pass

        # Force garbage collection
        gc.collect(2)


    def _process_files_streaming(self, s3_files, temp_dir, total_files, parallel_proc, decompile=None):
        """Process files in a streaming fashion using direct process management."""
        import queue

        # Use the instance's decompile flag if not provided
        if decompile is None:
            decompile = self.decompile

        # Use local tracking for statistics
        completed_count = 0
        skipped_count = 0
        failed_count = 0
        correctly_processed = 0
        partially_processed = 0
        filetype_stats = {}

        # Create a result queue for workers to return their results
        result_queue = multiprocessing.Queue()

        # Create a process ID tracking dict
        active_processes = {}  # {proc_id: (process, start_time, s3_key)}

        mode_str = "decompile" if decompile else "analysis"
        print(f"[INFO] Starting streaming processing of {len(s3_files)} files with {parallel_proc} workers ({mode_str} mode)")

        # Process files
        file_index = 0
        # Use different timeouts based on mode
        if decompile:
            worker_timeout = int(os.getenv("DECOMPILE_WORKER_TIMEOUT", "2700"))
        else:
            worker_timeout = int(os.getenv("REDB_TIMEOUT", "1200"))

        # Main processing loop
        while file_index < len(s3_files) or active_processes:
            # Start new processes if we have capacity and files to process
            while len(active_processes) < parallel_proc and file_index < len(s3_files):
                s3_bucket, s3_key, first_seen = s3_files[file_index]
                file_number = file_index + 1

                # Create and start a new process
                p = multiprocessing.Process(
                    target=direct_s3_worker,
                    args=(
                        s3_bucket,
                        s3_key,
                        temp_dir,
                        decompile,  # Pass the actual decompile flag
                        self.index_prefix,
                        self.log_file,
                        file_number,
                        total_files,
                        self.selected_modules,
                        result_queue,
                        self.dry_run,
                        self.yara_scan,
                        self.with_yara,
                        self.force,
                        self.decompile_modules,
                        first_seen,
                    )
                )
                p.start()

                # Track the process
                active_processes[p.pid] = (p, time.time(), s3_key, file_number)
                file_index += 1

                # Small delay to avoid overloading
                time.sleep(0.05)

            # Check for completed processes
            try:
                # Poll the result queue with a timeout
                while True:
                    try:
                        result = result_queue.get(block=True, timeout=1)

                        # Process result
                        s3_key = result.get('s3_key')
                        status = result.get('status', 'FAILED')
                        filetype = result.get('filetype')
                        file_number = result.get('file_number')
                        worker_pid = result.get('worker_pid')

                        # Remove from active processes if present
                        if worker_pid in active_processes:
                            del active_processes[worker_pid]

                        # Update statistics
                        if status == 'CORRECTLY':
                            correctly_processed += 1
                        elif status == 'PARTIALLY':
                            partially_processed += 1
                        elif status == 'SKIPPED':
                            skipped_count += 1
                        else:  # Any other status is treated as failure
                            failed_count += 1

                        if filetype:
                            filetype_stats[filetype] = filetype_stats.get(filetype, 0) + 1

                        # Update progress
                        completed_count += 1
                        if completed_count % 50 == 0 or completed_count == 1:
                            print(f"[INFO] Completed {completed_count}/{total_files} files. Last: {s3_key}")
                            print(f"[INFO] Progress - OK: {correctly_processed}, Partial: {partially_processed}, Failed: {failed_count}, Skipped: {skipped_count}, Active: {len(active_processes)}")

                    except queue.Empty:
                        # No results in the queue, break and check for timeouts
                        break

                # Check for timed-out processes
                current_time = time.time()
                timed_out_pids = []

                for pid, (proc, start_time, s3_key, file_number) in active_processes.items():
                    runtime = current_time - start_time

                    # Check if process has exceeded timeout
                    if runtime > worker_timeout:
                        print(f"[WARNING] Process {pid} processing {s3_key} exceeded timeout ({runtime:.0f}s > {worker_timeout}s)")

                        # Terminate the process
                        try:
                            proc.terminate()
                            time.sleep(0.1)  # Give it a moment to terminate
                            if proc.is_alive():
                                # If still alive, force kill
                                proc.kill()
                        except:
                            pass

                        # Clean up any child processes
                        try:
                            parent = psutil.Process(pid)
                            for child in parent.children(recursive=True):
                                try:
                                    child.kill()
                                except:
                                    pass
                        except:
                            pass

                        # Mark as failed
                        failed_count += 1
                        completed_count += 1
                        timed_out_pids.append(pid)

                    # Check if process has terminated without returning a result
                    elif not proc.is_alive():
                        print(f"[WARNING] Process {pid} processing {s3_key} terminated without result")

                        # Mark as failed
                        failed_count += 1
                        completed_count += 1
                        timed_out_pids.append(pid)

                # Remove timed-out processes from tracking
                for pid in timed_out_pids:
                    if pid in active_processes:
                        del active_processes[pid]

                # Sleep briefly to avoid hogging CPU
                time.sleep(0.1)

            except Exception as e:
                print(f"[ERROR] Exception in main processing loop: {e}")
                time.sleep(1)  # Sleep to avoid tight loop on error

        # Set final results in the shared dictionaries
        self.total_results[ImportResult.CORRECTLY] = correctly_processed
        self.total_results[ImportResult.PARTIALLY] = partially_processed
        self.total_results[ImportResult.FAILED] = failed_count
        self.total_results[ImportResult.SKIPPED] = skipped_count

        for filetype, count in filetype_stats.items():
            self.file_type_stats[filetype] = count

        print(f"[INFO] Streaming processing completed. Processed {completed_count}/{total_files} files.")

        # Final cleanup
        killed = self._kill_all_python_processes()
        if killed > 0:
            print(f"[INFO] Killed {killed} lingering processes during final cleanup")


    def _kill_all_python_processes(self):
        """Kill all python worker processes."""
        killed_count = 0
        for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
            try:
                if proc.info['name'] == 'python' and proc.info['cmdline']:
                    # Check if it's one of our processes
                    is_worker = False
                    for cmd in proc.info['cmdline']:
                        if 'deploy/redb/venv312bin/python' in cmd and 'multiprocessing' in cmd:
                            is_worker = True
                            break

                    if is_worker:
                        try:
                            proc.kill()
                            killed_count += 1
                        except Exception as e:
                            print(f"[ERROR] Failed to kill process {proc.info['pid']}: {e}")
            except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
                pass

        if killed_count > 0:
            print(f"[INFO] Killed {killed_count} python processes during cleanup")

        return killed_count


    def ingest(self):
        general_start_time = time.time()
        BATCH_SIZE = int(os.getenv("BATCH_SIZE", 1000))
        pool = None

        try:
            # Handle S3-solo mode
            if self.s3_solo:
                # Create a temporary directory for S3 downloads
                with tempfile.TemporaryDirectory() as temp_dir:
                    s3_bucket = os.getenv('S3_BUCKET')
                    if not s3_bucket:
                        print("[ERROR] S3_BUCKET environment variable is required for S3-solo mode")
                        return

                    # Use the provided S3 key directly (supports both sharded and private paths)
                    s3_key = self.s3_solo_key
                    print(f"[INFO] S3-solo mode: processing {s3_bucket}/{s3_key}")
                    print(f"[INFO] Extracted hash: {self.s3_solo_hash}")

                    # Fetch first_seen from catalog_samples
                    first_seen = None
                    try:
                        client = get_db_catalog_connection()
                        result = client.query(
                            "SELECT first_seen FROM catalog_samples WHERE sha256 = %(hash)s LIMIT 1",
                            parameters={"hash": self.s3_solo_hash}
                        )
                        if result.result_rows:
                            first_seen = result.result_rows[0][0]
                            print(f"[INFO] first_seen from catalog: {first_seen}")
                        else:
                            print(f"[INFO] No catalog entry found, first_seen will default to epoch zero")
                        client.close()
                    except Exception as e:
                        print(f"[WARNING] Could not fetch first_seen from catalog: {e}")

                    # Process single S3 file directly
                    result = process_s3_file(
                        s3_bucket,
                        s3_key,
                        temp_dir,
                        self.decompile,
                        self.index_prefix,
                        self.log_file,
                        1,  # file_number
                        1,  # total_files
                        selected_modules=self.selected_modules,
                        dry_run=self.dry_run,
                        yara_scan=self.yara_scan,
                        with_yara=self.with_yara,
                        force=self.force,
                        decompile_modules=self.decompile_modules,
                        first_seen=first_seen,
                    )

                    if result:
                        print(f"[INFO] S3-solo processing completed successfully")
                    else:
                        print(f"[ERROR] S3-solo processing failed")

                    return

            # Handle S3 bulk mode
            elif self.s3_mode:
                # Create a temporary directory for S3 downloads
                with tempfile.TemporaryDirectory() as temp_dir:
                    # Query catalog for S3 objects based on mode
                    if self.start_date and self.end_date:
                        # Date-based query: join catalog_samples with repository_upload_sessions
                        print(f"[INFO] Querying samples by date range: {self.start_date} to {self.end_date}")
                        # Pass repository as None if it's a default placeholder (not a real repo name)
                        repo_filter = self.repository if self.repository not in ("date-range", "analyzed") else None
                        catalog_entries = fetch_s3_objects_by_date_range(
                            index_prefix=self.index_prefix,
                            decompile=self.decompile,
                            start_date=self.start_date,
                            end_date=self.end_date,
                            repository=repo_filter,
                            notes=self.s3_notes,
                            magika_filter=self.magika_filter,
                            yara_scan=self.yara_scan,
                            force=self.force,
                            analyzed=self.analyzed
                        )
                        date_info = f" from {self.start_date} to {self.end_date}"
                    elif self.analyzed:
                        # Analyzed mode (standalone, no date filter): query basic_properties for already-analyzed samples
                        print(f"[INFO] Querying already-analyzed samples from {self.index_prefix}_basic_properties")
                        catalog_entries = fetch_analyzed_samples(
                            index_prefix=self.index_prefix,
                            decompile=self.decompile,
                            magika_filter=self.magika_filter,
                            yara_scan=self.yara_scan,
                            force=self.force,
                            rerun=self.rerun
                        )
                        date_info = ""
                    else:
                        # Repository-based query: direct query to repository_upload_sessions
                        catalog_entries = fetch_s3_objects_by_repository(
                            self.repository,
                            self.index_prefix,
                            self.decompile,
                            self.s3_notes,
                            self.magika_filter,
                            self.yara_scan,
                            self.force
                        )
                        date_info = ""

                    if not catalog_entries:
                        print(f"[INFO] No files found for repository: {self.repository}{date_info}" +
                              (f" with notes: {self.s3_notes}" if self.s3_notes else ""))
                        return

                    # Extract S3 bucket, keys, and first_seen
                    s3_files = []
                    for entry in catalog_entries:
                        s3_bucket = entry.get('s3_bucket')
                        s3_key = entry.get('s3_key')
                        first_seen = entry.get('first_seen')
                        if s3_bucket and s3_key:
                            s3_files.append((s3_bucket, s3_key, first_seen))

                    total_files = len(s3_files)
                    num_cores = multiprocessing.cpu_count()
                    parallel_proc = num_cores - 1

                    if self.decompile:
                        if self.magika_filter == 'apk':
                            parallel_proc = num_cores - 1
                            print(f"[INFO] Using decompile mode (APK) with {parallel_proc} parallel processes")
                        else:
                            parallel_proc = max(1, int((num_cores - 1) / 2))
                            print(f"[INFO] Using decompile mode with {parallel_proc} parallel processes")
                    else:
                        print(f"[INFO] Using analysis mode with {parallel_proc} parallel processes")

                    # Use streaming approach for both decompile and non-decompile cases
                    self._process_files_streaming(s3_files, temp_dir, total_files, parallel_proc, self.decompile)
            else:
                # Original file/directory processing logic
                if os.path.isfile(self.path):
                    # Check if it's a text file containing paths
                    if self.path.endswith('.txt'):
                        try:
                            with open(self.path, 'r') as f:
                                files = [line.strip() for line in f
                                    if line.strip() and not os.path.basename(line.strip()).startswith('.')]
                        except Exception as e:
                            print(f"[ERR] Failed to read file list from {self.path}: {str(e)}")
                            return
                    else:
                        files = [self.path]
                elif os.path.isdir(self.path):
                    files = [
                        os.path.join(root, file)
                        for root, dirs, files_list in os.walk(self.path)
                        for file in files_list
                        if not file.startswith('.')
                    ]
                else:
                    print(f"[ERR] Invalid path: {self.path}")
                    return

                total_files = len(files)
                num_cores = multiprocessing.cpu_count()
                parallel_proc = num_cores - 1
                if self.decompile:
                    if self.magika_filter == 'apk':
                        parallel_proc = num_cores - 1
                    else:
                        parallel_proc = int(num_cores/2)
                print(f"Number of CPU cores: {num_cores}")
                print(f"Number of parallel processes: {parallel_proc}")

                # Process files in batches
                for i in range(0, len(files), BATCH_SIZE):
                    batch_files = files[i:i + BATCH_SIZE]
                    batch_start = i
                    print(f"\nProcessing batch {i//BATCH_SIZE + 1}/{(len(files) + BATCH_SIZE - 1)//BATCH_SIZE}")
                    pool = None
                    try:
                        pool = Pool(processes=parallel_proc)
                        batch_results = pool.map(
                            worker,
                            [
                                (
                                    f,
                                    self.decompile,
                                    self.index_prefix,
                                    self.log_file,
                                    batch_start + idx + 1,
                                    total_files,
                                    self.selected_modules,
                                    self.dry_run,
                                    self.yara_scan,
                                    self.with_yara,
                                    self.force,
                                    self.decompile_modules,
                                )
                                for idx, f in enumerate(batch_files)
                            ],
                        )

                        # Update statistics for this batch
                        for result, filetype in batch_results:
                            if result is not None:
                                self.total_results[result] += 1
                            if filetype:
                                self.file_type_stats[filetype] = (
                                    self.file_type_stats.get(filetype, 0) + 1
                                )

                    finally:
                        # Properly close the pool after each batch
                        if pool:
                            try:
                                pool.close()
                                pool.join()
                                pool = None
                                gc.collect()
                            except Exception as e:
                                print(f"[ERROR] Error cleaning up pool: {e}")
                                try:
                                    pool.terminate()
                                    pool.join()
                                except:
                                    pass
                                pool = None
                                gc.collect()

                        if check_high_swap():
                            print("[INFO] High swap detected, restarting pool")
                            self.restart_worker_pool()

                            gc.collect(2)

                            # Create a fresh pool
                            self.pool = Pool(processes=parallel_proc)

                        # Export strings after each batch
            general_end_time = time.time()
            general_elapsed_time = general_end_time - general_start_time
            general_elapsed_time_pretty = time.strftime("%H:%M:%S", time.gmtime(general_elapsed_time))

            summary = (
                f"\n\nIngestion finished for {self.path}."
                f"\nTime required: {general_elapsed_time_pretty}"
                f"\nResults:"
                f"\n- Total analyzed: {sum(self.total_results.values())}"
                f"\n- Correctly imported: {self.total_results[ImportResult.CORRECTLY]}"
                f"\n- Partially imported: {self.total_results[ImportResult.PARTIALLY]}"
                f"\n- Failed: {self.total_results[ImportResult.FAILED]}"
                f"\n- Skipped: {self.total_results[ImportResult.SKIPPED]}"
                f"\n\nFiletype stats:\n{json.dumps(dict(self.file_type_stats))}\n"
            )

            with open(self.log_file, "a") as f:
                f.write(summary)

            print("Ingestion completed. Check the log file for details.")
            print(summary)

        finally:
            try:
                self._kill_all_python_processes()
            except Exception as e:
                print(f"[ERROR] Error in final cleanup: {e}")

            try:
                gc.collect(2)
            except Exception as e:
                print(f"[ERROR] Final garbage collection error: {e}")