<?php
/**
 * Niwa AI — Telegram album & text-fragment buffer (Phase 4 / feature/telegram-and-authonomy).
 *
 * Telegram delivers two kinds of "really one logical message" sequences as
 * multiple separate updates that we previously processed in isolation:
 *
 *   1. Albums: photos / videos sent together share a `media_group_id`. Each
 *      part is its own update with its own message_id but they belong to one
 *      logical "send 5 photos" action by the user. Without buffering, Niwa
 *      reacted to each photo as a separate turn, lost the album context, and
 *      sent the founder five replies for what they intended as one prompt.
 *
 *   2. Long-text fragments: Telegram clients hard-split messages over
 *      ~4096 chars into N updates with sequential message_ids fired within
 *      ~1.5 s. Without buffering, the agent saw a half-thought, replied to
 *      it, then saw the next half and replied again — confusing both sides.
 *
 * The buffer holds parts in WP options keyed by `media_group_id` /
 * `chat_id`, with a parallel index option (`OPT_AUTONOMY_TG_ALBUM_INDEX` /
 * `OPT_AUTONOMY_TG_TEXT_FRAG_INDEX`) because there is no "list all keys
 * with prefix" primitive. Options (rather than transients) are used so
 * the buffer survives across PHP-FPM workers regardless of how the host's
 * persistent object cache is configured — transients are cache-only when
 * `wp_using_ext_object_cache()` is true, which means a worker that buffers
 * a photo cannot reach a worker running the loopback flush if the cache
 * layer (LiteSpeed/Redis on managed hosts) is not shared between workers.
 * Manual `_expires_at` tracking replaces the transient TTL. Each poll cycle:
 *
 *   - calls `flush_ready_*` to drain entries that have aged past the idle
 *     threshold (or hit the hard TTL safety net), receiving combined
 *     payloads ready to feed back into `handle_incoming_message()`;
 *
 *   - pipes new updates through `buffer_*` to either grow an existing entry
 *     or start a fresh one (and short-circuits `handle_incoming_message`
 *     for buffered updates so they don't double-process).
 *
 * Modeled on Openclaw's `mediaGroupBuffer` and `textFragmentBuffer` in
 * `bot-handlers.ts` (`MEDIA_GROUP_TIMEOUT_MS`, `TELEGRAM_TEXT_FRAGMENT_*`)
 * but ported to PHP — there is no event loop or `setTimeout` in WP, so the
 * "did this entry idle out?" check happens cooperatively at every poll.
 *
 * @package Niwa_AI
 */

if ( ! defined( 'ABSPATH' ) ) {
    exit;
}

class Niwa_AI_Admin_Autonomy_Telegram_Album_Buffer {

    // Album: hard cap is intentionally generous so that very slow albums
    // (network jitter between parts) still combine. Idle threshold is what
    // actually decides when an album is "done".
    const ALBUM_TTL_SECONDS    = 30;
    const ALBUM_IDLE_FLUSH_MS  = 2500;

    // Long-text fragments: thresholds chosen to match Openclaw's defaults so
    // the user-perceived behavior is the same.
    const TEXT_FRAG_TTL_SECONDS              = 900;
    const TEXT_FRAG_GAP_MS                   = 1500;
    const TEXT_FRAG_MAX_PARTS                = 12;
    const TEXT_FRAG_MAX_TOTAL_CHARS          = 50000;
    const TEXT_FRAG_START_THRESHOLD_CHARS    = 4000;

    private $album_index_option;
    private $text_frag_index_option;
    private $album_transient_prefix    = 'niwa_ai_tg_album_';
    private $text_frag_transient_prefix = 'niwa_ai_tg_text_frag_';

    public function __construct() {
        $this->album_index_option     = Niwa_AI_Constants::OPT_AUTONOMY_TG_ALBUM_INDEX;
        $this->text_frag_index_option = Niwa_AI_Constants::OPT_AUTONOMY_TG_TEXT_FRAG_INDEX;
    }

    /* ───── Album buffering ──────────────────────────────────────────── */

    /**
     * Append a payload to the album buffer keyed by its `media_group_id`.
     * No-ops (returns false) when the payload is not part of an album.
     *
     * @param array $payload Output of Telegram::extract_message_payload().
     * @return bool True when buffered; false when the payload had no media_group_id.
     */
    public function buffer_album_part( $payload ) {
        $payload = is_array( $payload ) ? $payload : array();
        $mg      = isset( $payload['media_group_id'] ) ? (string) $payload['media_group_id'] : '';
        if ( '' === $mg ) {
            return false;
        }

        $key      = $this->album_transient_key( $mg );
        $existing = $this->read_buffer( $key );
        $now_ms   = $this->now_ms();

        if ( ! is_array( $existing ) ) {
            $existing = array(
                'media_group_id'      => $mg,
                'first_received_at_ms' => $now_ms,
                'last_received_at_ms'  => $now_ms,
                'parts'               => array(),
            );
            $this->add_to_index( $this->album_index_option, $mg );
        }

        $existing['last_received_at_ms'] = $now_ms;
        $existing['parts'][]             = $payload;

        $this->store_buffer( $key, $existing, self::ALBUM_TTL_SECONDS );
        return true;
    }

    /**
     * Return all album buffers that have aged past the idle threshold (or
     * the hard TTL), as combined payloads ready for handle_incoming_message().
     *
     * @return array<int, array>
     */
    public function flush_ready_albums() {
        $now_ms         = $this->now_ms();
        $idle_threshold = $now_ms - self::ALBUM_IDLE_FLUSH_MS;
        $hard_threshold = $now_ms - ( self::ALBUM_TTL_SECONDS * 1000 );

        $index     = (array) get_option( $this->album_index_option, array() );
        $ready     = array();
        $remaining = array();

        foreach ( $index as $mg ) {
            $mg    = (string) $mg;
            $key   = $this->album_transient_key( $mg );
            $entry = $this->read_buffer( $key );
            if ( ! is_array( $entry ) ) {
                // Buffer entry expired or was cleared externally; drop from index.
                continue;
            }

            $last_ms  = intval( $entry['last_received_at_ms'] ?? 0 );
            $first_ms = intval( $entry['first_received_at_ms'] ?? 0 );

            if ( $last_ms <= $idle_threshold || $first_ms <= $hard_threshold ) {
                $combined = $this->combine_album_parts( $entry );
                if ( null !== $combined ) {
                    $ready[] = $combined;
                }
                $this->delete_buffer( $key );
            } else {
                $remaining[] = $mg;
            }
        }

        update_option( $this->album_index_option, array_values( array_unique( $remaining ) ), false );
        return $ready;
    }

    /**
     * Force-flush a single album, regardless of its idle status.
     *
     * @param string $media_group_id
     * @return array|null Combined payload, or null if no buffer exists.
     */
    public function force_flush_album( $media_group_id ) {
        $mg    = (string) $media_group_id;
        $key   = $this->album_transient_key( $mg );
        $entry = $this->read_buffer( $key );
        if ( ! is_array( $entry ) ) {
            return null;
        }

        $this->delete_buffer( $key );
        $this->remove_from_index( $this->album_index_option, $mg );
        return $this->combine_album_parts( $entry );
    }

    /**
     * Merge an album buffer entry's parts into a single payload.
     *
     * Sort order: by `message_id` ASC so the resulting `media_refs[]` is in
     * the user's send order. The first part's identity (chat_id, from_*)
     * carries over; the highest `update_id` is preserved so offset tracking
     * remains correct after the combined payload is processed.
     *
     * @param array $entry Buffer entry.
     * @return array|null
     */
    private function combine_album_parts( $entry ) {
        $parts = isset( $entry['parts'] ) && is_array( $entry['parts'] ) ? $entry['parts'] : array();
        if ( empty( $parts ) ) {
            return null;
        }

        usort( $parts, function ( $a, $b ) {
            return intval( $a['message_id'] ?? 0 ) - intval( $b['message_id'] ?? 0 );
        } );

        $first              = $parts[0];
        $merged_media_refs  = array();
        $first_caption      = '';
        $max_update_id      = 0;
        $album_message_ids  = array();

        foreach ( $parts as $p ) {
            if ( ! empty( $p['media_refs'] ) && is_array( $p['media_refs'] ) ) {
                foreach ( $p['media_refs'] as $r ) {
                    $merged_media_refs[] = $r;
                }
            }
            $cap = isset( $p['caption'] ) ? (string) $p['caption'] : '';
            if ( '' !== $cap && '' === $first_caption ) {
                $first_caption = $cap;
            }
            $max_update_id      = max( $max_update_id, intval( $p['update_id'] ?? 0 ) );
            $album_message_ids[] = intval( $p['message_id'] ?? 0 );
        }

        $combined                       = $first;
        $combined['update_id']          = $max_update_id;
        $combined['media_refs']         = $merged_media_refs;
        $combined['caption']            = $first_caption;
        // BC text overload: text falls back to caption when text is empty.
        if ( '' === ( isset( $first['text_only'] ) ? (string) $first['text_only'] : '' ) ) {
            $combined['text'] = $first_caption;
        }
        $combined['has_user_text']      = '' !== ( isset( $combined['text_only'] ) ? (string) $combined['text_only'] : '' )
                                          || '' !== $first_caption;
        $combined['kind']               = 'album';
        $combined['placeholder']        = '<media:image>';
        $combined['is_album']           = true;
        $combined['album_size']         = count( $parts );
        $combined['album_message_ids']  = $album_message_ids;

        // reply_media_refs is per-message (only the first part typically carries
        // the reply context, since the user replied once and then attached the
        // album). Preserve from first part.
        $combined['reply_media_refs'] = isset( $first['reply_media_refs'] ) && is_array( $first['reply_media_refs'] )
            ? $first['reply_media_refs']
            : array();

        return $combined;
    }

    /* ───── Text-fragment buffering ──────────────────────────────────── */

    /**
     * Either append the payload to an existing in-flight fragment buffer for
     * this chat (when eligible) or start a new buffer if the text is at
     * threshold. Short text or non-eligible appends signal the caller to
     * process the payload normally.
     *
     * @param array $payload Telegram::extract_message_payload() output.
     * @return bool True when buffered; false when the caller should process normally.
     */
    public function buffer_text_fragment( $payload ) {
        $payload = is_array( $payload ) ? $payload : array();
        $chat_id = isset( $payload['chat_id'] ) ? (string) $payload['chat_id'] : '';
        if ( '' === $chat_id ) {
            return false;
        }

        // Phase 4 only buffers TEXT updates — anything with media_refs or a
        // caption is handled by the album buffer or processed normally so the
        // semantics stay clear. (Caption-bearing photos still go through
        // resolve_media_only_turn in Phase 5 routing.)
        if ( ! empty( $payload['media_refs'] ) ) {
            return false;
        }

        $text = isset( $payload['text_only'] ) ? (string) $payload['text_only'] : '';
        if ( '' === $text ) {
            return false;
        }

        $key      = $this->text_frag_transient_key( $chat_id );
        $existing = $this->read_buffer( $key );
        $now_ms   = $this->now_ms();

        if ( is_array( $existing ) && ! empty( $existing['parts'] ) ) {
            $last_part   = end( $existing['parts'] );
            $last_msg_id = intval( $last_part['message_id'] ?? 0 );
            $cur_msg_id  = intval( $payload['message_id'] ?? 0 );
            $time_gap_ms = $now_ms - intval( $existing['last_received_at_ms'] ?? 0 );
            $id_gap      = $cur_msg_id - $last_msg_id;

            $current_total = 0;
            foreach ( $existing['parts'] as $p ) {
                $current_total += strlen( (string) ( $p['text_only'] ?? '' ) );
            }
            $joiner        = isset( $existing['joiner'] ) ? (string) $existing['joiner'] : '';
            $current_total += strlen( $joiner ) * max( 0, count( $existing['parts'] ) - 1 );
            $next_total     = $current_total + strlen( $joiner ) + strlen( $text );

            $appendable = (
                $id_gap > 0 && $id_gap <= 1
                && $time_gap_ms <= self::TEXT_FRAG_GAP_MS
                && count( $existing['parts'] ) + 1 <= self::TEXT_FRAG_MAX_PARTS
                && $next_total <= self::TEXT_FRAG_MAX_TOTAL_CHARS
            );

            if ( $appendable ) {
                $existing['parts'][]             = $payload;
                $existing['last_received_at_ms'] = $now_ms;
                $this->store_buffer( $key, $existing, self::TEXT_FRAG_TTL_SECONDS );
                return true;
            }

            return false;
        }

        if ( strlen( $text ) < self::TEXT_FRAG_START_THRESHOLD_CHARS ) {
            return false;
        }

        // Only long Telegram-split text starts a fragment buffer. Normal
        // founder chat messages must enter the actor immediately.
        $entry = array(
            'chat_id'              => $chat_id,
            'first_received_at_ms' => $now_ms,
            'last_received_at_ms'  => $now_ms,
            'parts'                => array( $payload ),
            'joiner'               => "\n",
        );
        $this->store_buffer( $key, $entry, self::TEXT_FRAG_TTL_SECONDS );
        $this->add_to_index( $this->text_frag_index_option, $chat_id );
        return true;
    }

    /**
     * Drain text-fragment buffers that have aged past the gap threshold (or
     * the hard TTL).
     *
     * @return array<int, array>
     */
    public function flush_ready_text_fragments() {
        $now_ms         = $this->now_ms();
        $idle_threshold = $now_ms - self::TEXT_FRAG_GAP_MS;
        $hard_threshold = $now_ms - ( self::TEXT_FRAG_TTL_SECONDS * 1000 );

        $index     = (array) get_option( $this->text_frag_index_option, array() );
        $ready     = array();
        $remaining = array();

        foreach ( $index as $chat_id ) {
            $chat_id = (string) $chat_id;
            $key     = $this->text_frag_transient_key( $chat_id );
            $entry   = $this->read_buffer( $key );
            if ( ! is_array( $entry ) ) {
                continue;
            }

            $last_ms  = intval( $entry['last_received_at_ms'] ?? 0 );
            $first_ms = intval( $entry['first_received_at_ms'] ?? 0 );

            $entry_idle_threshold = $now_ms - self::TEXT_FRAG_GAP_MS;
            if ( $last_ms <= $entry_idle_threshold || $first_ms <= $hard_threshold ) {
                $combined = $this->combine_text_fragments( $entry );
                if ( null !== $combined ) {
                    $ready[] = $combined;
                }
                $this->delete_buffer( $key );
            } else {
                $remaining[] = $chat_id;
            }
        }

        update_option( $this->text_frag_index_option, array_values( array_unique( $remaining ) ), false );
        return $ready;
    }

    /**
     * Force-flush a chat's text-fragment buffer.
     *
     * @param string $chat_id
     * @return array|null
     */
    public function force_flush_text_fragment( $chat_id ) {
        $chat_id = (string) $chat_id;
        $key     = $this->text_frag_transient_key( $chat_id );
        $entry   = $this->read_buffer( $key );
        if ( ! is_array( $entry ) ) {
            return null;
        }

        $this->delete_buffer( $key );
        $this->remove_from_index( $this->text_frag_index_option, $chat_id );
        return $this->combine_text_fragments( $entry );
    }

    private function combine_text_fragments( $entry ) {
        $parts = isset( $entry['parts'] ) && is_array( $entry['parts'] ) ? $entry['parts'] : array();
        if ( empty( $parts ) ) {
            return null;
        }

        usort( $parts, function ( $a, $b ) {
            return intval( $a['message_id'] ?? 0 ) - intval( $b['message_id'] ?? 0 );
        } );

        $first             = $parts[0];
        $joiner            = isset( $entry['joiner'] ) ? (string) $entry['joiner'] : '';
        $combined_parts    = array();
        $max_update_id     = 0;
        $fragment_msg_ids  = array();
        foreach ( $parts as $p ) {
            $combined_parts[]   = (string) ( $p['text_only'] ?? '' );
            $max_update_id      = max( $max_update_id, intval( $p['update_id'] ?? 0 ) );
            $fragment_msg_ids[] = intval( $p['message_id'] ?? 0 );
        }
        $combined_text = implode( $joiner, $combined_parts );

        $combined                      = $first;
        $combined['update_id']         = $max_update_id;
        $combined['text']              = $combined_text;
        $combined['text_only']         = $combined_text;
        $combined['has_user_text']     = '' !== $combined_text;
        $combined['is_text_fragment']  = true;
        $combined['fragment_count']    = count( $parts );
        $combined['fragment_message_ids'] = $fragment_msg_ids;

        return $combined;
    }

    /* ───── Helpers ──────────────────────────────────────────────────── */

    private function now_ms() {
        return (int) round( microtime( true ) * 1000 );
    }

    private function album_transient_key( $media_group_id ) {
        return $this->album_transient_prefix . sha1( (string) $media_group_id );
    }

    private function text_frag_transient_key( $chat_id ) {
        return $this->text_frag_transient_prefix . sha1( (string) $chat_id );
    }

    private function add_to_index( $option_key, $value ) {
        $value = (string) $value;
        $index = (array) get_option( $option_key, array() );
        if ( ! in_array( $value, $index, true ) ) {
            $index[] = $value;
            update_option( $option_key, $index, false );
        }
    }

    private function remove_from_index( $option_key, $value ) {
        $value = (string) $value;
        $index = (array) get_option( $option_key, array() );
        $index = array_values( array_filter( $index, function ( $v ) use ( $value ) {
            return (string) $v !== $value;
        } ) );
        update_option( $option_key, $index, false );
    }

    /**
     * Persist a buffer entry to wp_options with a manual expiry. Options are
     * used instead of transients so the entry is reachable across PHP-FPM
     * workers regardless of the host's persistent object cache configuration —
     * transients live in the cache layer only when one is installed, which
     * means a worker that buffered a part cannot reach a worker running the
     * loopback flush if the cache is not shared between workers.
     */
    private function store_buffer( $key, $value, $ttl_seconds ) {
        $ttl_seconds = max( 1, intval( $ttl_seconds ) );
        update_option(
            $key,
            array(
                'value'      => $value,
                'expires_at' => time() + $ttl_seconds,
            ),
            false
        );
    }

    private function read_buffer( $key ) {
        $wrapped = get_option( $key, false );
        if ( ! is_array( $wrapped ) || ! array_key_exists( 'value', $wrapped ) ) {
            return false;
        }

        $expires_at = isset( $wrapped['expires_at'] ) ? intval( $wrapped['expires_at'] ) : 0;
        if ( $expires_at > 0 && $expires_at < time() ) {
            delete_option( $key );
            return false;
        }

        return $wrapped['value'];
    }

    private function delete_buffer( $key ) {
        delete_option( $key );
    }
}
