<?php
/**
 * SeekQuarry/Yioop --
 * Open Source Pure PHP Search Engine, Crawler, and Indexer
 *
 * Copyright (C) 2009 - 2026  Chris Pollett chris@pollett.org
 *
 * LICENSE:
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
 *
 * END LICENSE
 *
 * @author Chris Pollett chris@pollett.org
 * @license https://www.gnu.org/licenses/ GPL3
 * @link https://www.seekquarry.com/
 * @copyright 2009 - 2026
 * @filesource
 */
namespace seekquarry\yioop\library\storage_formats;
use seekquarry\yioop\library as L;

use seekquarry\yioop\configs as C;
use seekquarry\yioop\library\CrawlConstants;

/**  For remoteAddress and crawlHash function and Yioop constants */
require_once __DIR__ . "/../../library/Utility.php";
/**
 * Encapsulates the temporary storage of messages sent between QueueServers
 * and Fetchers during the course of a Crawl
 *
 * @author Chris Pollett
 */
class MessagesBundle implements CrawlConstants
{
    /**
    * Reference to a database object. Used since has directory manipulation
    * functions
    * @var object
     */
    public $db;
    /**
     * Largest sequence number currently in use for sliding window protocol
     * messages used for fetch batches of urls and downloaded web page
     * for those urls  responses
     * @var int
     */
    public $max_sequence_number;
    /**
     * Contains sequence_number => filename associations about which fetch
     * batches have been successfully  downloaded, received and should be
     * moved for further processing. These will be moved once all earlier
     * sequence numbers have been similarly received.
     * @var array
     */
    public $receive;
    /**
     * Contains sequence_number => [filename, schedule_time, tries] associations
     * for fetch batches of urls to crawl which have been produced. Here
     * filename is the name of the file that has the data for the batch,
     * schedule_time is the last time the batch was scheduled to a Fetcher,
     * and tries is the number of times it has been scheduled.
     * @var array
     */
    public $send;
    /**
     * String prefix to be used before filenames of message files in the
     * ETAG_EXPIRES_FOLDER, INDEX_DATA_FOLDER, ROBOTS_FOLDER, and
     * SCHEDULES_FOLDER folders.
     */
    const MESSAGE_PREFIX = "At";
    /**
     * Folder name for the folder within the MessageBundle folder used to
     * contain files with etag and expires header information waiting to be
     * processed by the scheduler
     */
    const ETAG_EXPIRES_FOLDER = "etag_expires";
    /**
    * Folder name for the folder within the MessageBundle folder used to
    * contain files with web page summaries and downloaded page information
    * waiting to be by the indexer (i.e., to make an inverted index)
    * processed
    */
    const INDEX_DATA_FOLDER = "index_data";
    /**
     * Folder name for the folder within the MessageBundle folder used to
     * contain files with robots.txt information waiting to be
     * processed by the scheduler
     */
    const ROBOTS_FOLDER = "robots";
    /**
    * Folder name for the folder within the MessageBundle folder used to
    * contain files with to crawl url information waiting to be
    * processed by the scheduler
    */
    const SCHEDULES_FOLDER = "schedules";
    /**
     * Makes a MessagesBundle with the provided parameters. This method
     * does not initialize send and receive properties related to
     * the fetch batch window.
     *
     * @param string $dir_name folder name used by this MessagesBundle
     * @param int $window_size number of outstanding url fetch batch
     *      files to allow before stop accepting new ones; used in a
     *      crawl to set the maximum number of schedules generated at
     *      a time. In the QueueServer the window size is set to be
     *      the total number of currently active fetchers
     */
    public function __construct(public $dir_name,
        public $window_size = 1)
    {
        if (!file_exists($dir_name)) {
            mkdir($dir_name);
            set_error_handler(null);
            @chmod($dir_name, 0777);
            restore_error_handler();
        }
        $db_class = C\NS_DATASOURCES . ucfirst(C\p('DBMS')) . "Manager";
        $this->db = new $db_class();
    }
    /**
     * Use to intialize the send and receiver property arrays for the
     * send and receive fetch batch window.
     *
     * @param bool $reset_schedule_times whether to reset the schedule
     * times of each batch current in the send folder, but not acknowledged
     * to 0, forcing them to eb rescheduled.
     */
    public function initWindows($reset_schedule_times = false)
    {
        $receive_dir = $this->dir_name . "/receive";
        $send_dir = $this->dir_name . "/send";
        $max_sequence_number = 0;
        foreach (['receive' => $receive_dir, 'send' => $send_dir] as
            $window_name => $dir) {
            if (!file_exists($dir)) {
                mkdir($dir);
                set_error_handler(null);
                @chmod($dir, 0777);
                restore_error_handler();
            }
            $files = glob("$dir/*.txt");
            $window_data = [];
            $sequence_number = 0;
            $len_prefix = strlen("$dir/");
            $len_suffix = strlen(".txt");
            foreach ($files as $file) {
                $base_name = substr($file, $len_prefix, -$len_suffix);
                $base_parts = explode("-", $base_name);
                $expected_num_parts = ($window_name == 'receive') ? 1 : 3;
                if (count($base_parts) != $expected_num_parts) {
                    continue;
                }
                $sequence_number = intval($base_parts[0]);
                if ($sequence_number > $max_sequence_number) {
                    $max_sequence_number = $sequence_number;
                }
                if ($window_name == 'send') {
                    list(, $schedule_time, $tries) = $base_parts;
                    $item = ['file' => $file,
                        'schedule_time' => intval($schedule_time),
                        'tries' => intval($tries)];
                    if ($reset_schedule_times) {
                        $schedule_time = 0;
                        $item['file'] = $this->makeSendWindowFilename(
                            $sequence_number, $schedule_time, $tries);
                        rename($file, $item['file']);
                    }
                } else {
                    $item = $file;
                }
                $window_data[$sequence_number] = $item;
            }
            $this->$window_name = $window_data;
        }
        $this->max_sequence_number = $max_sequence_number;
    }
    /**
     * Returns the data from the least sequence numbered file in the send
     * window that either has not previously been returned (tries 0) or
     * that has been previously returned but more than timeout seconds earlier
     * In both cases the try count of the file is bumped. If a file is
     * found in the search process with a try count above
     * MAX_RESCHEDULE_ATTEMPTS, it is moved instead to the receive window
     * but with no data received.
     *
     * @return string file contents of found file
     */
    public function extractSendWindow()
    {
        ksort($this->send);
        $data = false;
        $time = time();
        foreach($this->send as $sequence_number => $item) {
            $tries = $item['tries'];
            $timeout = ($item['schedule_time'] + C\SCHEDULE_TIMEOUT) < $time;
            if($item['tries'] == 0 || ($timeout &&
                $tries < C\MAX_RESCHEDULE_ATTEMPTS)) {
                $data = file_get_contents($item['file']);
                $tries++;
                $this->send[$sequence_number]['tries'] = $tries;
                $schedule_time = ($timeout) ? $time : $item['schedule_time'];
                $filename = $this->makeSendWindowFilename(
                    $sequence_number, $schedule_time, $tries);
                clearstatcache();
                if (file_exists($item['file'])) {
                    rename($item['file'], $filename);
                }
                $this->send[$sequence_number]['file'] = $filename;
                break;
            } else if ($timeout) {
                $this->updateReceiveWindow($sequence_number, "");
            }
        }
        return $data;
    }
    /**
     * Add a fetch batch to the send folder of the messages bundle if there
     * space in the folder given the current window size
     * @param string $tmp_send_filename filename of the file with the
     *  fetch batch. If this operation succeeds, the file at this location
     *  will be moved to the send folder
     * @return bool whether operation was successful
     */
    public function insertSendWindow($tmp_send_filename)
    {
        if (count($this->receive) + count($this->send) >=
            $this->window_size) {
            return false;
        }
        $schedule_time = time();
        $tries = 0;
        $this->max_sequence_number++;
        $sequence_number = $this->max_sequence_number;
        $filename = $this->makeSendWindowFilename($sequence_number,
            $schedule_time, $tries);
        rename($tmp_send_filename, $filename);
        set_error_handler(null);
        @chmod($filename, 0777);
        restore_error_handler();
        $this->send[$sequence_number] = ['file' => $filename,
            'schedule_time' => $schedule_time, 'tries' => $tries];
        return true;
    }
    /**
     * Used to output the filename of a file storing a fetch batch of urls in
     * the  send folder
     * @param int $sequence_number sequence number used for sliding window
     *  algorithm
     * @param int $schedule_time time at which the given file was last scheduled
     *  for download
     * @param int $tries number of times a fetcher has been scheduled to try
     *  to download the given list of urls
     * @return string the filename to use for the fetch batch with the given
     *  parameters
     */
    public function makeSendWindowFilename($sequence_number, $schedule_time,
        $tries)
    {
        return $this->dir_name .
            "/send/{$sequence_number}-$schedule_time-$tries.txt";
    }
    /**
     * Adds a new messages file to the received subfolder of the
     * MessagesBundle with the given sequence number and containing the
     * passed data
     *
     * @param int $sequence_number sequence number to use in making the
     *  filename of the new file to add to the receive folder
     * @param string $data message contents of the file to add
     */
    public function updateReceiveWindow($sequence_number, $data)
    {
        if (empty($this->send[$sequence_number])) {
            return;
        }
        $target_name = ($this->dir_name . "/receive/{$sequence_number}.txt");
        file_put_contents($target_name, $data);
        set_error_handler(null);
        @chmod($target_name, 0777);
        restore_error_handler();
        $this->receive[$sequence_number] = $target_name;
        unlink($this->send[$sequence_number]['file']);
        unset($this->send[$sequence_number]);
    }
    /**
     * Searches for a file is in the receive subfolder of this MessagesBundle
     * of least sequence number. If there is no file in the send subfolder
     * of the MessagesBundle with smaller sequence number, then the file
     * is read, deleted and its contents returned. Otherwise, this function
     * returns false.
     *
     * @return string|bool Data of found file or file
     */
    public function extractReceiveWindow()
    {
        ksort($this->send);
        $first_send_number = array_key_first($this->send);
        ksort($this->receive);
        $first_receive_number = array_key_first($this->receive);
        if ($first_send_number !== null &&
            $first_send_number < $first_receive_number) {
            return false;
        }
        $filename = $this->receive[$first_receive_number] ?? "";
        clearstatcache();
        if (!empty($filename) && file_exists($filename)) {
            $data = file_get_contents($filename);
            set_error_handler(null);
            @unlink($filename);
            restore_error_handler();
        } else {
            $data = false;
        }
        unset($this->receive[$first_receive_number]);
        return $data;
    }
    /**
     * Adds a file with contents $data and with name containing $address and
     * $time to a subfolder $day in the folder $messages_type subfolder of
     * the MessagesBundle
     *
     * @param string $messages_type the kind of messages being saved
     * @param string &$data_string encoded, compressed, serialized data the
     *     schedule is to contain
     * @param int $time Unix timestamp used to derive the day-bucket
     *     subfolder and the per-file timestamp; 0 substitutes time()
     */
    public function addMessages($messages_type, &$data_string, $time = 0)
    {
        $dir = $this->dir_name . "/$messages_type";
        $address = strtr(L\remoteAddress(), ["." => "-", ":" => "_"]);
        $time = ($time == 0) ? time() : $time;
        $day = floor($time/C\ONE_DAY);
        if (!file_exists($dir)) {
            mkdir($dir);
            chmod($dir, 0777);
        }
        $dir .= "/$day";
        if (!file_exists($dir)) {
            mkdir($dir);
            chmod($dir, 0777);
        }
        $data_hash = L\crawlHash($data_string);
        file_put_contents($dir . "/" . self::MESSAGE_PREFIX . $time .
            "From" . $address . "WithHash$data_hash.txt", $data_string);
    }
    /**
     * Get the next messages file by timestamp stored in the MessagesBundle
     * subfolder of of the given type (etag_expires, robots, schedules,
     * index_data) and either return it as a string or as a deserialized array
     * or object. This subfolder is typically organized into days subsubfolders
     * with the appropriately timestamped files in a given days folder.
     * After reading its data, but before returning it from
     * this method the file itself is deleted.
     *
     * @param string $type kind of data to look for messages about. This
     *  kind should correspond to a folder in the messages bundle. Typically,
     *  these folders will be (etag_expires, robots, schedules, index_data)
     * @param bool $unpack whether to unpack (decode/gzunzip/wddx_deserialize)
     *  the first found file and return its data, or just return its string
     *  contents after webdecode them.
     * @param string $message_prefix prefix string of filenames in
     *  given folder (etag_expires, robots, schedules, index_data) that have
     *  messages data
     * @return (string|array|object)?
     */
    public function nextMessages($type, $unpack = true, $message_prefix =
        self::MESSAGE_PREFIX)
    {
        $dirs = glob("{$this->dir_name}/$type/*", GLOB_ONLYDIR);
        $len_prefix =  strlen($message_prefix);
        foreach ($dirs as $dir) {
            $files = glob($dir . '/*.txt');
            if (isset($old_dir)) {
                L\crawlLog("Deleting $old_dir\n");
                $this->db->unlinkRecursive($old_dir);
                /* The idea is that only go through outer loop more than once
                   if earlier data directory empty.
                   Note: older directories should only have data dirs or
                   deleting like this might cause problems!
                 */
            }
            /* the code below returns the data
               from the lex-first file beginning with self::MESSAGE_PREFIX
             */
            foreach ($files as $file) {
                $path_parts = pathinfo($file);
                $base_name = $path_parts['basename'];
                $file_root_name = substr($base_name, 0, $len_prefix);
                if (strcmp($file_root_name, $message_prefix) == 0) {
                    if ($unpack) {
                        $message_data = $this->unpackMessages($file);
                    } else {
                        $message_data = L\webdecode(file_get_contents($file));
                    }
                    unlink($file);
                    return $message_data;
                }
            }
            $old_dir = $dir;
        }
        return null;
    }
    /**
     * Returns whether they are any as yet to be produced to crawl schedules
     * for the crawl queue in the given MessagesBundle's scheduler folder.
     * @param string $messages_bundle_dir folder to check from schedules
     * @return bool whether any found (true, if yes)
     */
    public static function isResumable($messages_bundle_dir)
    {
        $schedules_dir_name = "$messages_bundle_dir/" . self::SCHEDULES_FOLDER;
        if (!is_dir($schedules_dir_name)) {
            return false;
        }
        $schedules_dir = opendir($schedules_dir_name);
        $schedule_prefix = self::MESSAGE_PREFIX;
        $len_prefix = strlen(self::MESSAGE_PREFIX);
        while (($name = readdir($schedules_dir)) !==  false) {
            $sub_path = "$schedules_dir_name/$name";
            if (!is_dir($sub_path) || $name == '.' ||
                $name == '..') {
                continue;
            }
            $sub_dir = opendir($sub_path);
            while (($sub_name = readdir($sub_dir)) !== false) {
                if (substr($sub_name, 0, $len_prefix) == $schedule_prefix) {
                    return true;
                }
            }
            closedir($sub_dir);
        }
        closedir($schedules_dir);
        return false;
    }
    /**
     * Decodes the web-string safety encoding and then gunzips and deserializes
     * the result for a messages file used for communication between a Fetcher
     * and the QueueServer
     * @param string $file name of the
     * @return array|object unserialized data array or object contained in the
     *  messages file to be processed
     */
    public function unpackMessages($file)
    {
        L\crawlLog("Processing File: $file");
        $decode = file_get_contents($file);
        $decode = L\webdecode($decode);
        set_error_handler(null);
        $decode = @gzuncompress($decode);
        $sites = @unserialize($decode);
        restore_error_handler();
        return $sites;
    }
}
