meshcore-bot/modules/feed_manager.py

#!/usr/bin/env python3
"""
Feed Manager for RSS and API feed subscriptions
Handles polling feeds and sending updates to channels
"""

import asyncio
import contextlib
import hashlib
import json
import os
import sqlite3
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional
from urllib.parse import urlparse

import aiohttp
import feedparser

from modules.feed_filter_eval import (
    get_nested_value,
    item_passes_filter_config,
    parse_microsoft_date,
)
from modules.feed_format import (
    apply_feed_field_function,
    feed_format_auto_base_value,
    feed_format_auto_slots,
    format_feed_message,
    format_relative_timestamp,
    sort_feed_items,
    truncate_to_budget,
)
from modules.security_utils import (
    SafeAiohttpResolver,
    SafeUrlPolicy,
    UnsafeUrlError,
    safe_aiohttp_request,
)

DEFAULT_MAX_FEED_RESPONSE_BYTES = 2 * 1024 * 1024
DEFAULT_MAX_PARSED_FEED_ITEMS = 500


def _useful_feed_content_type(content_type: str, feed_type: str) -> bool:
    """Accept common feed/API media types while rejecting clearly unrelated bodies."""
    media_type = content_type.partition(';')[0].strip().lower()
    if not media_type:
        return True
    shared_fallbacks = {'text/plain', 'application/octet-stream'}
    if media_type in shared_fallbacks:
        return True
    if feed_type == 'rss':
        return (
            media_type in {
                'application/rss+xml',
                'application/atom+xml',
                'application/xml',
                'text/xml',
                'application/xhtml+xml',
                'text/html',
            }
            or media_type.endswith('+xml')
        )
    return media_type in {'application/json', 'text/json'} or media_type.endswith('+json')


class FeedManager:
    """Manages RSS and API feed subscriptions"""

    def __init__(self, bot):
        self.bot = bot
        self.logger = bot.logger
        self.db_path = bot.db_manager.db_path

        # Configuration (guard against missing [Feed_Manager] section for upgrade compatibility)
        if not bot.config.has_section('Feed_Manager'):
            self.enabled = False
            self.default_check_interval = 300
            self.max_items_per_check = 10
            self.max_posts_per_check = 10
            self.request_timeout = 30
            self.user_agent = 'MeshCoreBot/1.0 FeedManager'
            self.rate_limit_seconds = 5.0
            self.max_message_length = 130
            self.default_output_format = '{emoji} {body|truncate:100} - {date}\n{link|truncate:50}'
            self.default_send_interval = 2.0
            self.max_response_bytes = DEFAULT_MAX_FEED_RESPONSE_BYTES
            self.max_parsed_items = DEFAULT_MAX_PARSED_FEED_ITEMS
            self.shorten_feed_urls = False
            if bot.config.has_section('Feed_Command'):
                try:
                    self.allow_private_urls = bot.config.getboolean(
                        'Feed_Command',
                        'allow_private_urls',
                        fallback=False,
                    )
                except ValueError:
                    self.allow_private_urls = False
            else:
                self.allow_private_urls = False
        else:
            self.enabled = bot.config.getboolean('Feed_Manager', 'feed_manager_enabled', fallback=False)
            self.default_check_interval = bot.config.getint('Feed_Manager', 'default_check_interval_seconds', fallback=300)
            # Clamped to >= 1: the scan window feeds a list slice, where 0 disables
            # posting entirely and negatives take Python's negative-slice meaning
            # (drop the newest N items) — neither is what a config typo intends.
            self.max_items_per_check = max(
                1, bot.config.getint('Feed_Manager', 'max_items_per_check', fallback=10)
            )
            # Max items actually posted per check. Defaults to max_items_per_check so existing
            # installs behave identically; raise max_items_per_check (the scan window) to reach
            # older passing items while this caps how many post per poll. Also clamped to >= 1,
            # since the post loop only checks the cap after sending an item.
            self.max_posts_per_check = max(
                1,
                bot.config.getint(
                    'Feed_Manager', 'max_posts_per_check', fallback=self.max_items_per_check
                ),
            )
            # Clamped to >= 1: aiohttp.ClientTimeout only arms a timer for
            # total > 0, so 0 or negative disables the timeout entirely and a
            # hung feed server holds its poll lock and semaphore slot forever.
            self.request_timeout = max(
                1, bot.config.getint('Feed_Manager', 'feed_request_timeout', fallback=30)
            )
            self.user_agent = bot.config.get('Feed_Manager', 'feed_user_agent', fallback='MeshCoreBot/1.0 FeedManager')
            self.rate_limit_seconds = bot.config.getfloat('Feed_Manager', 'feed_rate_limit_seconds', fallback=5.0)
            # Clamped: the final truncation is `message[:max_message_length - 3]`,
            # so anything under 4 becomes a negative slice that lengthens the
            # message instead of capping it.
            self.max_message_length = max(
                10, bot.config.getint('Feed_Manager', 'max_message_length', fallback=130)
            )
            self.default_output_format = bot.config.get('Feed_Manager', 'default_output_format', fallback='{emoji} {body|truncate:100} - {date}\n{link|truncate:50}')
            self.default_send_interval = bot.config.getfloat('Feed_Manager', 'default_message_send_interval_seconds', fallback=2.0)
            self.max_response_bytes = max(
                1024,
                bot.config.getint(
                    'Feed_Manager',
                    'max_response_bytes',
                    fallback=DEFAULT_MAX_FEED_RESPONSE_BYTES,
                ),
            )
            self.max_parsed_items = max(
                1,
                bot.config.getint(
                    'Feed_Manager',
                    'max_parsed_items',
                    fallback=DEFAULT_MAX_PARSED_FEED_ITEMS,
                ),
            )
            self.shorten_feed_urls = bot.config.getboolean(
                'Feed_Manager', 'shorten_urls', fallback=False
            )
            if bot.config.has_section('Feed_Command'):
                try:
                    feed_command_allow_private = bot.config.getboolean(
                        'Feed_Command',
                        'allow_private_urls',
                        fallback=False,
                    )
                except ValueError:
                    feed_command_allow_private = False
            else:
                feed_command_allow_private = False
            self.allow_private_urls = bot.config.getboolean(
                'Feed_Manager',
                'allow_private_urls',
                fallback=feed_command_allow_private,
            )

        # Rate limiting per domain
        self._domain_last_request: dict[str, float] = {}
        self._domain_rate_locks: dict[str, asyncio.Lock] = {}
        self._feed_poll_locks: dict[int, asyncio.Lock] = {}

        # HTTP session
        self.session: Optional[aiohttp.ClientSession] = None
        self._url_policy = SafeUrlPolicy(allow_private=self.allow_private_urls)

        # Semaphore to limit concurrent requests
        self._request_semaphore = asyncio.Semaphore(5)

        # Serialize process_message_queue; lock is checked before acquiring to avoid coroutine pileup
        self._process_queue_lock: Optional[asyncio.Lock] = None
        # Persisted across runs so per-feed send intervals are respected without sleeping under the lock
        self._feed_last_send: dict[int, float] = {}

        self.logger.info("FeedManager initialized")

    async def initialize(self):
        """Initialize the feed manager (create HTTP session)"""
        if not self.enabled:
            self.logger.info("FeedManager is disabled in config")
            return

        # Don't create session here - create it lazily when needed
        # This avoids issues with using sessions across different event loops
        # The session will be created in the same event loop where it's used
        self.logger.info("FeedManager initialized (session will be created on first use)")

    async def stop(self):
        """Stop the feed manager (close HTTP session)"""
        if self.session and not self.session.closed:
            await self.session.close()
            self.session = None
        self.logger.info("FeedManager stopped")

    async def poll_all_feeds(self):
        """Poll all enabled feeds that are due for checking"""
        if not self.enabled:
            return

        try:
            # Get all enabled feeds
            feeds = self._get_enabled_feeds()

            if not feeds:
                return

            # Filter feeds that are due for checking
            current_time = time.time()
            feeds_to_check = []

            for feed in feeds:
                last_check = feed.get('last_check_time')
                if last_check:
                    try:
                        # Parse timestamp - handle both ISO format and SQLite format
                        if isinstance(last_check, str):
                            # Try ISO format first (with timezone)
                            try:
                                last_check_dt = datetime.fromisoformat(last_check.replace('Z', '+00:00'))
                            except ValueError:
                                # Try SQLite format (YYYY-MM-DD HH:MM:SS) - treat as UTC
                                try:
                                    last_check_dt = datetime.strptime(last_check, '%Y-%m-%d %H:%M:%S')
                                    last_check_dt = last_check_dt.replace(tzinfo=timezone.utc)
                                except ValueError:
                                    # Try with microseconds
                                    try:
                                        last_check_dt = datetime.strptime(last_check, '%Y-%m-%d %H:%M:%S.%f')
                                        last_check_dt = last_check_dt.replace(tzinfo=timezone.utc)
                                    except ValueError:
                                        raise ValueError(f"Unknown timestamp format: {last_check}")
                        else:
                            last_check_dt = datetime.fromtimestamp(last_check, tz=timezone.utc)

                        # Convert to timestamp
                        if last_check_dt.tzinfo:
                            last_check_ts = last_check_dt.timestamp()
                        else:
                            # Assume UTC if no timezone
                            last_check_ts = last_check_dt.replace(tzinfo=timezone.utc).timestamp()
                    except Exception as e:
                        self.logger.debug(f"Error parsing last_check_time for feed {feed['id']}: {e}")
                        last_check_ts = 0
                else:
                    last_check_ts = 0

                # Fall back for rows that predate interval validation: NULL would
                # raise a TypeError below and abort the poll cycle for *every*
                # feed, and <= 0 marks the feed permanently due.
                interval = feed.get('check_interval_seconds')
                if not isinstance(interval, (int, float)) or interval <= 0:
                    interval = self.default_check_interval

                if current_time - last_check_ts >= interval:
                    feeds_to_check.append(feed)

            if not feeds_to_check:
                self.logger.debug("No feeds due for checking at this time")
                return

            self.logger.info(f"Polling {len(feeds_to_check)} feed(s) that are due for checking")

            # Poll feeds in parallel (with semaphore limit)
            tasks = [self.poll_feed(feed) for feed in feeds_to_check]
            await asyncio.gather(*tasks, return_exceptions=True)

        except Exception as e:
            self.logger.error(f"Error in poll_all_feeds: {e}")

    async def _ensure_session(self):
        """Ensure HTTP session exists in the current event loop"""
        if self.session is None or self.session.closed:
            # Create session in the current event loop context
            self.session = aiohttp.ClientSession(
                headers={'User-Agent': self.user_agent},
                connector=aiohttp.TCPConnector(
                    resolver=SafeAiohttpResolver(self._url_policy),
                    use_dns_cache=False,
                ),
            )
            self.logger.debug("Created FeedManager HTTP session in current event loop")

    async def _read_limited_response(
        self,
        response: aiohttp.ClientResponse,
        feed_type: str,
    ) -> bytes:
        """Read a decompressed response body without exceeding the configured cap."""
        content_type = response.headers.get('Content-Type', '')
        if not _useful_feed_content_type(content_type, feed_type):
            raise ValueError(f"Unexpected {feed_type.upper()} content type: {content_type}")

        declared_length = response.headers.get('Content-Length')
        if declared_length:
            try:
                content_length = int(declared_length)
            except ValueError:
                content_length = None
            if content_length is not None and content_length > self.max_response_bytes:
                raise ValueError(
                    f"Feed response exceeds {self.max_response_bytes} byte limit"
                )

        chunks: list[bytes] = []
        total = 0
        async for chunk in response.content.iter_chunked(64 * 1024):
            total += len(chunk)
            if total > self.max_response_bytes:
                raise ValueError(
                    f"Feed response exceeds {self.max_response_bytes} byte limit"
                )
            chunks.append(chunk)
        return b''.join(chunks)

    async def poll_feed(self, feed: dict[str, Any]):
        """Poll a single feed, serializing concurrent attempts for that feed."""
        feed_id = int(feed['id'])
        lock = self._feed_poll_locks.setdefault(feed_id, asyncio.Lock())
        async with lock:
            await self._poll_feed_locked(feed)

    async def _poll_feed_locked(self, feed: dict[str, Any]) -> None:
        """Poll a feed while its per-feed exclusion lock is held."""
        # Ensure session exists in current event loop
        await self._ensure_session()

        feed_id = feed['id']
        feed_type = feed['feed_type']
        feed_url = feed['feed_url']
        feed['channel_name']

        try:
            # Validate URL for SSRF protection
            try:
                await self._url_policy.validate_async(feed_url)
            except UnsafeUrlError:
                self.logger.error(f"Feed URL validation failed: {feed_url}")
                self._record_feed_error(feed_id, 'security', 'Invalid or unsafe URL')
                return

            self.logger.debug(f"Polling {feed_type} feed {feed_id}: {feed_url}")

            # Rate limit per domain
            host = self._normalized_host(feed_url)
            await self._wait_for_rate_limit(host)

            # Fetch feed data
            if feed_type == 'rss':
                new_items = await self.process_rss_feed(feed)
            elif feed_type == 'api':
                new_items = await self.process_api_feed(feed)
            else:
                self.logger.warning(f"Unknown feed type: {feed_type}")
                return

            # Process new items.
            # Examine up to max_items_per_check items (the scan window) and post the ones that
            # pass the filter, stopping once max_posts_per_check items have been queued. This
            # keeps filtered-out items from consuming the post budget, so a long back-catalog
            # with a restrictive filter (e.g. within_days) doesn't stall behind old items.
            if new_items:
                self.logger.info(f"Found {len(new_items)} new items for feed {feed_id}")
                filtered_count = 0
                posted_count = 0
                for item in new_items[:self.max_items_per_check]:
                    # Budget checked before sending, not after: checking after meant
                    # a cap of 0 still sent one item. __init__ clamps the config
                    # value, but the attribute is public and set directly in tests
                    # and by callers, so guard the loop itself.
                    if posted_count >= self.max_posts_per_check:
                        break
                    # Check if item passes filter conditions
                    if self._should_send_item(feed, item):
                        await self._send_feed_item(feed, item)
                        posted_count += 1
                    else:
                        filtered_count += 1
                        self.logger.debug(f"Filtered out item: {item.get('title', 'Untitled')[:50]}")

                if filtered_count > 0:
                    self.logger.debug(f"Filtered out {filtered_count} items for feed {feed_id}")
            else:
                self.logger.debug(f"No new items found for feed {feed_id}")

            # Always update last check time, even if no new items
            self._update_feed_last_check(feed_id)

        except Exception as e:
            self.logger.error(f"Error polling feed {feed_id}: {e}")
            self._record_feed_error(feed_id, 'network', str(e))

    async def process_rss_feed(self, feed: dict[str, Any]) -> list[dict[str, Any]]:
        """Process an RSS feed and return new items"""
        feed_url = feed['feed_url']
        last_item_id = feed.get('last_item_id')

        try:
            # Fetch RSS feed - use aiohttp's timeout directly
            # Create timeout object in the current async context
            timeout = aiohttp.ClientTimeout(total=self.request_timeout)

            async with self._request_semaphore:
                try:
                    assert self.session is not None
                    response = await safe_aiohttp_request(
                        self.session,
                        "GET",
                        feed_url,
                        policy=self._url_policy,
                        timeout=timeout,
                    )
                    try:
                        if response.status != 200:
                            raise Exception(f"HTTP {response.status}")
                        content = await self._read_limited_response(response, 'rss')
                    finally:
                        response.release()
                except (asyncio.TimeoutError, aiohttp.ServerTimeoutError):
                    raise Exception(f"Request timeout after {self.request_timeout} seconds")

            # Parse RSS feed
            parsed = feedparser.parse(content)

            if parsed.bozo:
                self.logger.warning(f"RSS feed parsing warning: {parsed.bozo_exception}")

            # Extract items - collect ALL items first (don't break early if sorting is configured)
            all_items = []
            for entry in parsed.entries[:self.max_parsed_items]:
                # Get item ID (prefer guid, then link, then hash of title+link)
                item_id = entry.get('id') or entry.get('guid') or entry.get('link')
                if not item_id:
                    # Generate ID from title and link
                    item_id = hashlib.md5(
                        f"{entry.get('title', '')}{entry.get('link', '')}".encode()
                    ).hexdigest()

                # Parse published date
                published = None
                if hasattr(entry, 'published_parsed') and entry.published_parsed:
                    with contextlib.suppress(Exception):
                        published = datetime(*entry.published_parsed[:6], tzinfo=timezone.utc)

                all_items.append({
                    'id': item_id,
                    'title': entry.get('title', 'Untitled'),
                    'link': entry.get('link', ''),
                    'description': entry.get('description', ''),
                    'published': published
                })

            # Apply sorting if configured (before filtering, so we can properly track the last item)
            sort_config_str = feed.get('sort_config')
            if sort_config_str:
                try:
                    sort_config = json.loads(sort_config_str) if isinstance(sort_config_str, str) else sort_config_str
                    all_items = self._sort_items(all_items, sort_config)
                except (json.JSONDecodeError, TypeError, Exception) as e:
                    self.logger.warning(f"Error applying sort config for feed {feed['id']}: {e}")

            # Reverse to get oldest first (if no sort config)
            if not sort_config_str:
                all_items.reverse()

            # Now filter out items that have already been processed
            # Check against both last_item_id and the feed_activity table for robust deduplication
            items = []
            processed_item_ids = set()

            # Get all previously processed item IDs from feed_activity table
            if last_item_id:
                processed_item_ids.add(last_item_id)

            # Query database for all processed item IDs for this feed
            try:
                with self.bot.db_manager.connection() as conn:
                    cursor = conn.cursor()
                    cursor.execute('''
                        SELECT item_id FROM feed_activity WHERE feed_id = ?
                        UNION
                        SELECT item_id FROM feed_message_queue
                        WHERE feed_id = ? AND sent_at IS NULL
                          AND item_id IS NOT NULL AND trim(item_id) <> ''
                    ''', (feed['id'], feed['id']))
                    for row in cursor.fetchall():
                        processed_item_ids.add(row[0])
            except Exception as e:
                self.logger.warning(f"Error querying processed items for feed {feed['id']}: {e}")

            # Filter out already processed items
            for item in all_items:
                if item['id'] not in processed_item_ids:
                    items.append(item)
                else:
                    self.logger.debug(f"Skipping already processed item {item['id']} for feed {feed['id']}")

            # Update last_item_id if we have new items (use the last item from the sorted list)
            if items:
                # Use the last item from the original sorted list (all_items), not the filtered list
                # This ensures we track the most recent item even if it was already processed
                self._update_feed_last_item_id(feed['id'], all_items[-1]['id'])

            return items

        except Exception as e:
            self.logger.error(f"Error processing RSS feed: {e}")
            raise

    async def process_api_feed(self, feed: dict[str, Any]) -> list[dict[str, Any]]:
        """Process an API feed and return new items"""
        feed_url = feed['feed_url']
        api_config_str = feed.get('api_config', '{}')
        last_item_id = feed.get('last_item_id')

        try:
            # Parse API config
            api_config = json.loads(api_config_str) if api_config_str else {}

            method = api_config.get('method', 'GET').upper()
            headers = api_config.get('headers', {})
            params = api_config.get('params', {})
            body = api_config.get('body')
            parser_config = api_config.get('response_parser', {})

            # Make HTTP request - use aiohttp's timeout directly
            # Create timeout object in the current async context
            timeout = aiohttp.ClientTimeout(total=self.request_timeout)

            async with self._request_semaphore:
                try:
                    assert self.session is not None
                    response = await safe_aiohttp_request(
                        self.session,
                        method,
                        feed_url,
                        policy=self._url_policy,
                        headers=headers,
                        params=params,
                        json=body if method == 'POST' else None,
                        timeout=timeout,
                    )
                    try:
                        if response.status != 200:
                            raise Exception(f"HTTP {response.status}")
                        content = await self._read_limited_response(response, 'api')
                        data = json.loads(content)
                    finally:
                        response.release()
                except (asyncio.TimeoutError, aiohttp.ServerTimeoutError):
                    raise Exception(f"Request timeout after {self.request_timeout} seconds")

            # Extract items using parser config
            items_path = parser_config.get('items_path', '')
            if items_path:
                # Navigate JSON path
                parts = items_path.split('.')
                items_data = data
                for part in parts:
                    items_data = items_data.get(part, [])
            else:
                # Assume data is a list
                items_data = data if isinstance(data, list) else [data]

            # Extract items
            id_field = parser_config.get('id_field', 'id')
            title_field = parser_config.get('title_field', 'title')
            description_field = parser_config.get('description_field', 'description')  # New: allow custom description field
            timestamp_field = parser_config.get('timestamp_field', 'created_at')
            emoji_field = parser_config.get('emoji_field', 'emoji')  # New: allow custom per-item emoji field

            # Collect ALL items first (don't break early, as sorting may reorder them)
            all_items = []
            if not isinstance(items_data, list):
                items_data = [items_data]
            for item_data in items_data[:self.max_parsed_items]:
                if not isinstance(item_data, dict):
                    continue
                item_id = str(self._get_nested_value(item_data, id_field, ''))
                if not item_id:
                    continue

                # Parse timestamp if available - support nested paths
                published = None
                if timestamp_field:
                    ts_value = self._get_nested_value(item_data, timestamp_field)
                    if ts_value:
                        try:
                            if isinstance(ts_value, (int, float)):
                                published = datetime.fromtimestamp(ts_value, tz=timezone.utc)
                            elif isinstance(ts_value, str):
                                # Try Microsoft date format first
                                if ts_value.startswith('/Date('):
                                    published = self._parse_microsoft_date(ts_value)
                                else:
                                    # Try ISO format
                                    try:
                                        published = datetime.fromisoformat(ts_value.replace('Z', '+00:00'))
                                    except ValueError:
                                        # Try common formats
                                        for fmt in ['%Y-%m-%dT%H:%M:%S', '%Y-%m-%d %H:%M:%S', '%Y-%m-%d']:
                                            try:
                                                published = datetime.strptime(ts_value, fmt)
                                                if published.tzinfo is None:
                                                    published = published.replace(tzinfo=timezone.utc)
                                                break
                                            except ValueError:
                                                continue
                        except Exception:
                            pass

                # Get description - support nested paths
                description = ''
                if description_field:
                    desc_value = self._get_nested_value(item_data, description_field)
                    if desc_value:
                        description = str(desc_value)

                all_items.append({
                    'id': item_id,
                    'title': self._get_nested_value(item_data, title_field, 'Untitled'),
                    'emoji': self._get_nested_value(item_data, emoji_field, ''),
                    'link': item_data.get('link', ''),
                    'description': description,
                    'published': published,
                    'raw': item_data  # Store full raw response for field access
                })

            # Apply sorting if configured (before filtering, so we can properly track the last item)
            sort_config_str = feed.get('sort_config')
            if sort_config_str:
                try:
                    sort_config = json.loads(sort_config_str) if isinstance(sort_config_str, str) else sort_config_str
                    all_items = self._sort_items(all_items, sort_config)
                except (json.JSONDecodeError, TypeError, Exception) as e:
                    self.logger.warning(f"Error applying sort config for feed {feed['id']}: {e}")

            # Reverse to get oldest first (if no sort config)
            if not sort_config_str:
                all_items.reverse()

            # Now filter out items that have already been processed
            # Check against both last_item_id and the feed_activity table for robust deduplication
            items = []
            processed_item_ids = set()

            # Get all previously processed item IDs from feed_activity table
            if last_item_id:
                processed_item_ids.add(last_item_id)

            # Query database for all processed item IDs for this feed
            try:
                with self.bot.db_manager.connection() as conn:
                    cursor = conn.cursor()
                    cursor.execute('''
                        SELECT item_id FROM feed_activity WHERE feed_id = ?
                        UNION
                        SELECT item_id FROM feed_message_queue
                        WHERE feed_id = ? AND sent_at IS NULL
                          AND item_id IS NOT NULL AND trim(item_id) <> ''
                    ''', (feed['id'], feed['id']))
                    for row in cursor.fetchall():
                        processed_item_ids.add(row[0])
            except Exception as e:
                self.logger.warning(f"Error querying processed items for feed {feed['id']}: {e}")

            # Filter out already processed items
            for item in all_items:
                if item['id'] not in processed_item_ids:
                    items.append(item)
                else:
                    self.logger.debug(f"Skipping already processed item {item['id']} for feed {feed['id']}")

            # Update last_item_id if we have new items (use the last item from the sorted list)
            if items:
                # Use the last item from the original sorted list (all_items), not the filtered list
                # This ensures we track the most recent item even if it was already processed
                self._update_feed_last_item_id(feed['id'], all_items[-1]['id'])

            return items

        except Exception as e:
            self.logger.error(f"Error processing API feed: {e}")
            raise

    def _format_timestamp(self, published: Optional[datetime]) -> str:
        """Format a timestamp as a relative time string"""
        return format_relative_timestamp(published)

    @staticmethod
    def _feed_format_auto_slots(format_str: str) -> list[tuple[int, int, str]]:
        """Return (start, end, field_name) for each {field|auto} placeholder (left-to-right)."""
        return feed_format_auto_slots(format_str)

    @staticmethod
    def _truncate_to_budget(text: str, budget: int) -> str:
        """Fit text to at most budget characters; ellipsis when budget > 3 (same idea as truncate:N)."""
        return truncate_to_budget(text, budget)

    def _feed_format_auto_base_value(
        self,
        field_name: str,
        raw_data: Any,
        replacements: dict[str, str],
        link_original: str,
    ) -> str:
        """Full string for one field before |auto (long link, no shorten)."""
        return feed_format_auto_base_value(
            field_name, raw_data, replacements, link_original
        )

    def _apply_shortening(self, text: str, function: str) -> str:
        """Apply a shortening, parsing, or conditional function to text."""
        return apply_feed_field_function(
            text, function, config=self.bot.config, logger=self.logger
        )

    def _get_nested_value(self, data: Any, path: str, default: Any = '') -> Any:
        """Get a nested value from a dict/list using dot notation."""
        return get_nested_value(data, path, default)

    def _parse_microsoft_date(self, date_str: str) -> Optional[datetime]:
        """Parse Microsoft JSON date format: /Date(timestamp-offset)/"""
        return parse_microsoft_date(date_str)

    def _sort_items(self, items: list[dict[str, Any]], sort_config: dict) -> list[dict[str, Any]]:
        """Sort items based on sort configuration."""
        return sort_feed_items(
            items, sort_config, log_warning=self.logger.warning
        )

    def format_message(self, item: dict[str, Any], feed: dict[str, Any]) -> str:
        """Format a feed item as a message for the mesh using configurable format with placeholders.

        Supported placeholders and field functions are documented on
        modules.feed_format.format_feed_message.
        """
        format_str = feed.get('output_format') or self.default_output_format
        return format_feed_message(
            item,
            format_str,
            feed_name=feed.get('feed_name') or '',
            feed_id=feed.get('id'),
            max_message_length=self.max_message_length,
            shorten_feed_urls=self.shorten_feed_urls,
            config=self.bot.config,
            logger=self.logger,
        )

    def _queue_feed_message(
        self,
        feed: dict[str, Any],
        item: dict[str, Any],
        message: str,
    ) -> bool:
        """Queue a feed message, returning whether a new row was inserted."""
        try:
            with self.bot.db_manager.connection() as conn:
                cursor = conn.cursor()
                cursor.execute('''
                    INSERT INTO feed_message_queue
                    (feed_id, channel_name, message, item_id, item_title, priority)
                    VALUES (?, ?, ?, ?, ?, 0)
                    ON CONFLICT(feed_id, item_id)
                    WHERE item_id IS NOT NULL AND trim(item_id) <> ''
                    DO NOTHING
                ''', (
                    feed['id'],
                    feed['channel_name'],
                    message,
                    item.get('id', ''),
                    item.get('title', '')[:200]  # Limit title length
                ))
                conn.commit()
                inserted = cursor.rowcount == 1
                if inserted:
                    self.logger.debug(f"Queued feed message for {feed['channel_name']}: {item.get('title', '')[:50]}")
                else:
                    self.logger.debug(
                        f"Skipped duplicate queued item {item.get('id', '')!r} "
                        f"for feed {feed['id']}"
                    )
                return inserted
        except Exception as e:
            self.logger.error(f"Error queuing feed message: {e}")
            self._record_feed_error(feed['id'], 'queue', str(e))
            return False

    def _should_send_item(self, feed: dict[str, Any], item: dict[str, Any]) -> bool:
        """Check if an item should be sent based on filter configuration.

        See modules/feed_filter_eval.py and docs/FEEDS.md for operators.
        """
        def _warn(msg: str) -> None:
            self.logger.warning(f"{msg} (feed id {feed.get('id')})")

        return item_passes_filter_config(
            item,
            feed.get('filter_config'),
            log_warning=_warn,
        )

    async def _send_feed_item(self, feed: dict[str, Any], item: dict[str, Any]):
        """Queue a feed item message instead of sending immediately"""
        try:
            message = self.format_message(item, feed)
            # Queue the message instead of sending immediately
            self._queue_feed_message(feed, item, message)
        except Exception as e:
            self.logger.error(f"Error processing feed item: {e}")
            self._record_feed_error(feed['id'], 'other', str(e))

    @staticmethod
    def _normalized_host(url: str) -> str:
        """Return a stable host key independent of case, port, or trailing dot."""
        host = (urlparse(url).hostname or '').rstrip('.').lower()
        try:
            return host.encode('idna').decode('ascii')
        except UnicodeError:
            return host

    async def _wait_for_rate_limit(self, host: str):
        """Serialize and space requests to the same normalized host."""
        lock = self._domain_rate_locks.setdefault(host, asyncio.Lock())
        async with lock:
            last_request = self._domain_last_request.get(host)
            if last_request is not None:
                elapsed = time.monotonic() - last_request
                if elapsed < self.rate_limit_seconds:
                    await asyncio.sleep(self.rate_limit_seconds - elapsed)
            self._domain_last_request[host] = time.monotonic()

    def _get_enabled_feeds(self) -> list[dict[str, Any]]:
        """Get all enabled feed subscriptions from database"""
        try:
            with self.bot.db_manager.connection() as conn:
                conn.row_factory = sqlite3.Row
                cursor = conn.cursor()
                cursor.execute('''
                    SELECT * FROM feed_subscriptions
                    WHERE enabled = 1
                    ORDER BY last_check_time ASC NULLS FIRST
                ''')
                rows = cursor.fetchall()
                return [dict(row) for row in rows]
        except Exception as e:
            self.logger.error(f"Error getting enabled feeds: {e}")
            return []

    def _update_feed_last_check(self, feed_id: int):
        """Update the last check time for a feed"""
        try:
            from datetime import datetime, timezone
            # Use Python's datetime to ensure proper timezone handling
            # Store in ISO format with timezone for JavaScript compatibility
            now = datetime.now(timezone.utc)
            now_str = now.isoformat()  # ISO format: 2025-12-05T12:34:56.789+00:00

            with self.bot.db_manager.connection() as conn:
                cursor = conn.cursor()
                cursor.execute('''
                    UPDATE feed_subscriptions
                    SET last_check_time = ?,
                        updated_at = ?
                    WHERE id = ?
                ''', (now_str, now_str, feed_id))
                conn.commit()
                self.logger.debug(f"Updated last_check_time for feed {feed_id} to {now_str}")
        except Exception as e:
            self.logger.error(f"Error updating feed last check: {e}")

    def _update_feed_last_item_id(self, feed_id: int, item_id: str):
        """Update the last processed item ID for a feed"""
        try:
            with self.bot.db_manager.connection() as conn:
                cursor = conn.cursor()
                cursor.execute('''
                    UPDATE feed_subscriptions
                    SET last_item_id = ?,
                        updated_at = CURRENT_TIMESTAMP
                    WHERE id = ?
                ''', (item_id, feed_id))
                conn.commit()
        except Exception as e:
            self.logger.error(f"Error updating feed last item ID: {e}")

    def _record_feed_activity(self, feed_id: int, item_id: str, item_title: str):
        """Record that a feed item was processed"""
        try:
            with self.bot.db_manager.connection() as conn:
                cursor = conn.cursor()
                cursor.execute('''
                    INSERT INTO feed_activity (feed_id, item_id, item_title, message_sent)
                    VALUES (?, ?, ?, 1)
                ''', (feed_id, item_id, item_title[:200]))  # Limit title length
                conn.commit()
        except Exception as e:
            self.logger.error(f"Error recording feed activity: {e}")

    def _record_feed_error(self, feed_id: int, error_type: str, error_message: str):
        """Record a feed error"""
        try:
            with self.bot.db_manager.connection() as conn:
                cursor = conn.cursor()
                cursor.execute('''
                    INSERT INTO feed_errors (feed_id, error_type, error_message)
                    VALUES (?, ?, ?)
                ''', (feed_id, error_type, error_message[:500]))  # Limit message length
                conn.commit()
        except Exception as e:
            self.logger.error(f"Error recording feed error: {e}")

    async def process_message_queue(self):
        """Process queued feed messages and send them at configured intervals"""
        if self._process_queue_lock is None:
            self._process_queue_lock = asyncio.Lock()
        if self._process_queue_lock.locked():
            return  # previous run still in progress; skip this tick
        async with self._process_queue_lock:
            await self._process_message_queue_inner()

    async def _process_message_queue_inner(self):
        """Body of process_message_queue (runs under _process_queue_lock)."""
        try:
            # Get all unsent messages, ordered by priority and queue time
            db_path = str(self.db_path)  # Ensure string, not Path object
            with self.bot.db_manager.connection() as conn:
                conn.row_factory = sqlite3.Row
                cursor = conn.cursor()
                cursor.execute('''
                    SELECT q.id, q.feed_id, q.channel_name, q.message, q.item_id, q.item_title,
                           f.message_send_interval_seconds
                    FROM feed_message_queue q
                    JOIN feed_subscriptions f ON q.feed_id = f.id
                    WHERE q.sent_at IS NULL
                    ORDER BY q.priority DESC, q.queued_at ASC
                    LIMIT 100
                ''')
                messages = cursor.fetchall()

            if not messages:
                return

            for msg in messages:
                feed_id = msg['feed_id']
                channel_name = msg['channel_name']
                message_text = msg['message']
                queue_id = msg['id']
                item_id = msg['item_id']
                item_title = msg['item_title']

                # Get send interval for this feed (default if not set)
                send_interval = msg['message_send_interval_seconds'] or self.default_send_interval

                # Skip messages whose feed interval hasn't elapsed yet; the next
                # scheduler tick (2 s) will retry without blocking under the lock.
                if feed_id in self._feed_last_send:
                    elapsed = time.time() - self._feed_last_send[feed_id]
                    if elapsed < send_interval:
                        continue

                # Send the message
                try:
                    success = await self.bot.command_manager.send_channel_message(channel_name, message_text)

                    if success:
                        # Mark as sent
                        with self.bot.db_manager.connection() as conn:
                            cursor = conn.cursor()
                            cursor.execute('''
                                UPDATE feed_message_queue
                                SET sent_at = CURRENT_TIMESTAMP
                                WHERE id = ?
                            ''', (queue_id,))
                            conn.commit()

                        # Record activity
                        self._record_feed_activity(feed_id, item_id, item_title)
                        self.logger.debug(f"Sent queued feed message to {channel_name}: {item_title[:50]}")
                        self._feed_last_send[feed_id] = time.time()
                    else:
                        self.logger.warning(f"Failed to send queued feed message to channel {channel_name}")
                        self._record_feed_error(feed_id, 'channel', f"Failed to send to channel {channel_name}")
                        # Don't mark as sent, will retry later

                except Exception as e:
                    self.logger.error(f"Error sending queued feed message: {e}")
                    self._record_feed_error(feed_id, 'other', str(e))
                    # Don't mark as sent, will retry later

        except Exception as e:
            db_path = getattr(self, 'db_path', 'unknown')
            db_path_str = str(db_path) if db_path != 'unknown' else 'unknown'
            self.logger.exception(f"Error processing message queue: {e}")
            if db_path_str != 'unknown':
                path_obj = Path(db_path_str)
                self.logger.error(f"Database path: {db_path_str} (exists: {path_obj.exists()}, readable: {os.access(db_path_str, os.R_OK) if path_obj.exists() else False}, writable: {os.access(db_path_str, os.W_OK) if path_obj.exists() else False})")
                # Check parent directory permissions
                if path_obj.exists():
                    parent = path_obj.parent
                    self.logger.error(f"Parent directory: {parent} (exists: {parent.exists()}, writable: {os.access(str(parent), os.W_OK) if parent.exists() else False})")
            else:
                self.logger.error(f"Database path: {db_path_str}")