893 lines
58 KiB
Markdown
893 lines
58 KiB
Markdown
# Data Flow Documentation
|
|
|
|
Complete documentation of data flows through the RedditBot system.
|
|
|
|
---
|
|
|
|
## Table of Contents
|
|
|
|
1. [Overview](#overview)
|
|
2. [Post Processing Flow](#post-processing-flow)
|
|
3. [Response Monitoring Flow](#response-monitoring-flow)
|
|
4. [Chat Interaction Flow](#chat-interaction-flow)
|
|
5. [Strategy Selection Flow](#strategy-selection-flow)
|
|
6. [Media Sync Flow](#media-sync-flow)
|
|
7. [Auto-Reply Flow](#auto-reply-flow)
|
|
8. [Dashboard Data Flow](#dashboard-data-flow)
|
|
9. [Error Recovery Flows](#error-recovery-flows)
|
|
10. [Data Lifecycle](#data-lifecycle)
|
|
|
|
---
|
|
|
|
## Overview
|
|
|
|
RedditBot processes data through several interconnected pipelines:
|
|
|
|
```
|
|
┌─────────────┐ ┌──────────────┐ ┌─────────────┐
|
|
│ Reddit API │────►│ Monitor │────►│ Processor │
|
|
│ (Posts) │ │ Service │ │ Service │
|
|
└─────────────┘ └──────────────┘ └──────┬──────┘
|
|
│
|
|
┌──────────────┐ ▼
|
|
│ Response │ ┌──────────────┐
|
|
│ Checker │◄────│ OpenRouter │
|
|
└──────────────┘ │ (AI) │
|
|
└──────────────┘
|
|
│
|
|
┌──────────────┐ ▼
|
|
│ Chat │ ┌──────────────┐
|
|
│ Service │◄────│ Message │
|
|
└──────────────┘ │ Sender │
|
|
└──────────────┘
|
|
│
|
|
┌──────────────┐ ▼
|
|
│ Media │ ┌──────────────┐
|
|
│ Service │◄────│ SQLite DB │
|
|
└──────────────┘ └──────────────┘
|
|
│
|
|
┌──────────────┐ │
|
|
│ Dashboard │◄──────────┘
|
|
│ (Express) │
|
|
└──────────────┘
|
|
```
|
|
|
|
---
|
|
|
|
## Post Processing Flow
|
|
|
|
The primary data flow for discovering, analyzing, and messaging about new posts.
|
|
|
|
### Stage 1: Discovery
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ MonitorService │
|
|
│ │
|
|
│ 1. Shuffle subreddit order │
|
|
│ 2. For each subreddit: │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ RedditApi.fetchPosts(subreddit) │ │
|
|
│ │ │ │
|
|
│ │ HTTP: GET /r/{subreddit}/new.json?limit=100 │ │
|
|
│ │ │ │
|
|
│ │ Response: { │ │
|
|
│ │ data: { │ │
|
|
│ │ children: [ │ │
|
|
│ │ { │ │
|
|
│ │ data: { │ │
|
|
│ │ id: "abc123", │ │
|
|
│ │ subreddit: "gonewild", │ │
|
|
│ │ author: "user123", │ │
|
|
│ │ title: "Summer vibes", │ │
|
|
│ │ url: "https://...", │ │
|
|
│ │ preview: { reddits: [...] }, │ │
|
|
│ │ thumbnail: "https://...", │ │
|
|
│ │ post_hint: "image", │ │
|
|
│ │ over_18: true │ │
|
|
│ │ } │ │
|
|
│ │ } │ │
|
|
│ │ ] │ │
|
|
│ │ } │ │
|
|
│ │ } │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 3. Filter posts: │
|
|
│ ├─► post_hint === "image" │
|
|
│ ├─► author exists and is not deleted │
|
|
│ ├─► image_url can be extracted │
|
|
│ └─► Author not in seller/spam list │
|
|
│ │
|
|
│ 4. Cross-reference with DB: │
|
|
│ ├─► PostsRepository.getById(postId) → skip if exists │
|
|
│ └─► MessagesRepository.getByAuthor(author) → skip if sent │
|
|
│ │
|
|
│ 5. Return newest valid post │
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
### Stage 2: Image Processing
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ ProcessorService │
|
|
│ │
|
|
│ 1. Download Image │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ ImageService.downloadImage(url) │ │
|
|
│ │ │ │
|
|
│ │ HTTP: GET {image_url} │ │
|
|
│ │ │ │
|
|
│ │ Response: Binary image data │ │
|
|
│ │ │ │
|
|
│ │ Validation: │ │
|
|
│ │ ├─► MIME type: image/jpeg | png | webp │ │
|
|
│ │ ├─► File size: < 10MB │ │
|
|
│ │ └─► Not empty/broken │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 2. Convert to base64 │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ Buffer.toString('base64') │ │
|
|
│ │ │ │
|
|
│ │ Result: "data:image/jpeg;base64,/9j/4AAQSkZJR..." │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
### Stage 3: Gender Detection (Pre-Filter)
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ OpenRouterService │
|
|
│ │
|
|
│ 1. detectGenderOnly(imageBase64) │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ HTTP: POST https://openrouter.ai/api/v1/chat/ │ │
|
|
│ │ completions │ │
|
|
│ │ │ │
|
|
│ │ Request: { │ │
|
|
│ │ model: "fast-model", │ │
|
|
│ │ messages: [ │ │
|
|
│ │ { │ │
|
|
│ │ role: "user", │ │
|
|
│ │ content: [ │ │
|
|
│ │ { type: "text", text: "Is this F or M?" }, │ │
|
|
│ │ { type: "image_url", image_url: {...} } │ │
|
|
│ │ ] │ │
|
|
│ │ } │ │
|
|
│ │ ] │ │
|
|
│ │ } │ │
|
|
│ │ │ │
|
|
│ │ Response: { │ │
|
|
│ │ choices: [{ message: { content: "F" } }] │ │
|
|
│ │ } │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 3. Compare with config.targetGender │
|
|
│ ├─► Match: continue to full analysis │
|
|
│ └─► No match: SKIP post (log: "Gender mismatch, skipping") │
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
### Stage 4: Strategy Selection
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ StrategySelector │
|
|
│ │
|
|
│ 1. Get strategy statistics from DB │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ MessagesRepository.getStrategyStats() │ │
|
|
│ │ │ │
|
|
│ │ Result: { │ │
|
|
│ │ variant_a: { sent: 50, responded: 15, rate: 0.30 },│
|
|
│ │ variant_b: { sent: 30, responded: 12, rate: 0.40 },│
|
|
│ │ variant_c: { sent: 5, responded: 1, rate: 0.20 } │
|
|
│ │ } │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 2. Apply exploration/exploit algorithm │
|
|
│ ├─► Random roll < explorationRate → pick random variant │
|
|
│ └─► Otherwise → pick highest response rate (min samples) │
|
|
│ │
|
|
│ 3. Return selected variant name │
|
|
│ Example: "variant_b" │
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
### Stage 5: Full AI Analysis
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ OpenRouterService │
|
|
│ │
|
|
│ 1. analyzeImage(imageBase64) │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ HTTP: POST https://openrouter.ai/api/v1/chat/ │ │
|
|
│ │ completions │ │
|
|
│ │ │ │
|
|
│ │ Request: { │ │
|
|
│ │ model: "primary-model", │ │
|
|
│ │ messages: [ │ │
|
|
│ │ { │ │
|
|
│ │ role: "system", │ │
|
|
│ │ content: config.openrouter.systemPrompt │ │
|
|
│ │ }, │ │
|
|
│ │ { │ │
|
|
│ │ role: "user", │ │
|
|
│ │ content: [ │ │
|
|
│ │ { type: "text", text: "Describe this image" },│
|
|
│ │ { type: "image_url", image_url: {...} } │ │
|
|
│ │ ] │ │
|
|
│ │ } │ │
|
|
│ │ ] │ │
|
|
│ │ } │ │
|
|
│ │ │ │
|
|
│ │ Response: { │ │
|
|
│ │ choices: [{ │ │
|
|
│ │ message: { │ │
|
|
│ │ content: '{"gender":"F","description":"..."}' │ │
|
|
│ │ } │ │
|
|
│ │ }] │ │
|
|
│ │ } │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 2. Check for refusal │
|
|
│ ├─► If refused → try fallback models │
|
|
│ └─► If all failed → mark post with error, skip │
|
|
│ │
|
|
│ 3. Parse and return ImageAnalysis │
|
|
│ { gender: "F", description: "Young woman, blonde hair..." }│
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
### Stage 6: Message Generation
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ OpenRouterService │
|
|
│ │
|
|
│ 1. Retrieve few-shot examples │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ SuccessfulMessagesRepository.getExamples( │ │
|
|
│ │ subreddit, count: 3 │ │
|
|
│ │ ) │ │
|
|
│ │ │ │
|
|
│ │ Result: [ │ │
|
|
│ │ { │ │
|
|
│ │ original_message: "Hey, you look amazing...", │ │
|
|
│ │ response_snippet: "Thank you so much!", │ │
|
|
│ │ subreddit: "gonewild" │ │
|
|
│ │ }, │ │
|
|
│ │ // ... more examples │ │
|
|
│ │ ] │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 2. Generate message │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ HTTP: POST https://openrouter.ai/api/v1/chat/ │ │
|
|
│ │ completions │ │
|
|
│ │ │ │
|
|
│ │ Request: { │ │
|
|
│ │ model: "primary-model", │ │
|
|
│ │ messages: [ │ │
|
|
│ │ { role: "system", content: messagePrompt }, │ │
|
|
│ │ { role: "user", content: [ │ │
|
|
│ │ `Image: ${imageAnalysis.description}`, │ │
|
|
│ │ `Post title: ${postTitle}`, │ │
|
|
│ │ `Strategy: ${selectedVariant}`, │ │
|
|
│ │ `Examples: ${JSON.stringify(examples)}` │ │
|
|
│ │ ] │ │
|
|
│ │ } │ │
|
|
│ │ ] │ │
|
|
│ │ } │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 3. Return generated message │
|
|
│ "Hey, I came across your post and..." │
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
### Stage 7: DM Sending
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ MessagePage │
|
|
│ │
|
|
│ 1. Navigate to compose page │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ page.goto("https://www.reddit.com/compose?to= │ │
|
|
│ │ user123") │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 2. Fill recipient field │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ page.evaluate(() => { │ │
|
|
│ │ const input = document.querySelector( │ │
|
|
│ │ 'faceplate-autocomplete input' │ │
|
|
│ │ ); │ │
|
|
│ │ input.value = "user123"; │ │
|
|
│ │ input.dispatchEvent(new Event('input', │ │
|
|
│ │ { bubbles: true })); │ │
|
|
│ │ }); │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 3. Fill subject field ("Stunning") │
|
|
│ 4. Fill body field (generated message) │
|
|
│ 5. Click send button │
|
|
│ 6. Verify success (check for confirmation) │
|
|
│ │
|
|
│ Result: true (sent) or false (failed) │
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
### Stage 8: Database Recording
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ Database │
|
|
│ │
|
|
│ 1. INSERT INTO processed_posts │
|
|
│ VALUES ( │
|
|
│ 'abc123', -- post_id │
|
|
│ 'gonewild', -- subreddit │
|
|
│ 'user123', -- author │
|
|
│ 'Summer vibes', -- title │
|
|
│ 'https://...', -- image_url │
|
|
│ '{"gender":"F",...}', -- image_analysis │
|
|
│ 'Hey, I came across...', -- generated_message │
|
|
│ 1716234567, -- processed_at │
|
|
│ 1, -- message_sent │
|
|
│ NULL -- error │
|
|
│ ); │
|
|
│ │
|
|
│ 2. INSERT INTO sent_messages │
|
|
│ VALUES ( │
|
|
│ NULL, -- id (auto) │
|
|
│ 'abc123', -- post_id │
|
|
│ 'user123', -- recipient │
|
|
│ 'Stunning', -- subject │
|
|
│ 'Hey, I came across...', -- message │
|
|
│ 1716234567, -- sent_at │
|
|
│ 'variant_b', -- strategy_id │
|
|
│ 0, - - got_response │
|
|
│ NULL -- response_checked_at │
|
|
│ ); │
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
---
|
|
|
|
## Response Monitoring Flow
|
|
|
|
How the bot detects and records user replies to sent DMs.
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ ResponseCheckerService │
|
|
│ │
|
|
│ 1. Navigate to inbox │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ page.goto("https://www.reddit.com/message/inbox") │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 2. Parse inbox items │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ Extract from page: │ │
|
|
│ │ [ │ │
|
|
│ │ { │ │
|
|
│ │ from: "user456", │ │
|
|
│ │ subject: "Re: Stunning", │ │
|
|
│ │ snippet: "Thank you so much! 😊", │ │
|
|
│ │ timestamp: 1716234000, │ │
|
|
│ │ link: "/message/xyz789" │ │
|
|
│ │ }, │ │
|
|
│ │ // ... more messages │ │
|
|
│ │ ] │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 3. Match responses to sent messages │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ For each inbox item: │ │
|
|
│ │ ├─► MessagesRepository.getByRecipient(from) │ │
|
|
│ │ ├─► Find unmatched message within time window │ │
|
|
│ │ ├─► Update: got_response = 1 │ │
|
|
│ │ ├─► Update: response_checked_at = now │ │
|
|
│ │ └─► Store in successful_messages │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 4. Update strategy statistics │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ MessagesRepository.updateStrategyStats( │ │
|
|
│ │ strategy_id, got_response: true │ │
|
|
│ │ ) │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 5. Log results │
|
|
│ └─► "Found 3 responses, updated 2 strategies" │
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
---
|
|
|
|
## Chat Interaction Flow
|
|
|
|
How the bot interacts with Reddit chat conversations.
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ ChatService │
|
|
│ │
|
|
│ ──── PHASE 1: FETCH CHATS ────────────────────────────────── │
|
|
│ │
|
|
│ 1. Navigate to chat page │
|
|
│ page.goto("https://chat.reddit.com") │
|
|
│ │
|
|
│ 2. Traverse shadow DOM for conversations │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ page.evaluate(() => { │ │
|
|
│ │ // JS injection to pierce shadow DOM │ │
|
|
│ │ const roots = document.querySelectorAll( │ │
|
|
│ │ 'faceplate-chat-list' │ │
|
|
│ │ ); │ │
|
|
│ │ const chats = []; │ │
|
|
│ │ roots.forEach(root => { │ │
|
|
│ │ const shadow = root.shadowRoot; │ │
|
|
│ │ const items = shadow.querySelectorAll( │ │
|
|
│ │ '.chat-item' │ │
|
|
│ │ ); │ │
|
|
│ │ items.forEach(item => { │ │
|
|
│ │ chats.push({ │ │
|
|
│ │ id: item.dataset.chatId, │ │
|
|
│ │ participant: item.querySelector( │ │
|
|
│ │ '.username' │ │
|
|
│ │ ).textContent, │ │
|
|
│ │ lastMessage: item.querySelector( │ │
|
|
│ │ '.message' │ │
|
|
│ │ ).textContent, │ │
|
|
│ │ timestamp: item.dataset.timestamp │ │
|
|
│ │ }); │ │
|
|
│ │ }); │ │
|
|
│ │ }); │ │
|
|
│ │ return chats; │ │
|
|
│ │ }); │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 3. Cache results │
|
|
│ └─► Save to memory + disk JSON │
|
|
│ │
|
|
│ ──── PHASE 2: LOAD CONVERSATION ──────────────────────────── │
|
|
│ │
|
|
│ 1. Check cache │
|
|
│ ├─► If cached and fresh: return from memory │
|
|
│ └─► If not: fetch from disk or Reddit │
|
|
│ │
|
|
│ 2. Fetch full history (pagination) │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ For each page of messages: │ │
|
|
│ │ ├─► Navigate to chat view │ │
|
|
│ │ ├─► Extract messages from shadow DOM │ │
|
|
│ │ ├─► Scroll to load more (infinite scroll) │ │
|
|
│ │ └─► Repeat until all loaded │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 3. Return conversation object │
|
|
│ { │
|
|
│ chatId: "abc123", │
|
|
│ participants: ["user456", "bot"], │
|
|
│ messages: [ │
|
|
│ { author: "user456", text: "Hey!", timestamp: ... }, │
|
|
│ { author: "bot", text: "Hi there!", timestamp: ... }, │
|
|
│ // ... more messages │
|
|
│ ], │
|
|
│ lastUpdated: 1716234567 │
|
|
│ } │
|
|
│ │
|
|
│ ──── PHASE 3: SEND MESSAGE ───────────────────────────────── │
|
|
│ │
|
|
│ 1. Navigate to chat view │
|
|
│ 2. Fill message input (shadow DOM) │
|
|
│ 3. Apply typing delay (human-like) │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ const typingDelay = random(500, 2000); │ │
|
|
│ │ await page.waitForTimeout(typingDelay); │ │
|
|
│ │ │ │
|
|
│ │ // Simulate typing character by character │ │
|
|
│ │ for (const char of message) { │ │
|
|
│ │ input.value += char; │ │
|
|
│ │ await page.waitForTimeout(random(30, 100)); │ │
|
|
│ │ } │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ 4. Submit message │
|
|
│ 5. Verify delivery (check for sent indicator) │
|
|
│ │
|
|
│ Result: true (delivered) or false (failed) │
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
---
|
|
|
|
## Strategy Selection Flow
|
|
|
|
How A/B test variants are selected and updated.
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ StrategySelector │
|
|
│ │
|
|
│ SELECT STRATEGY │
|
|
│ ──────────────────────────────────────────────────────────── │
|
|
│ │
|
|
│ Input: subreddit, author │
|
|
│ │
|
|
│ 1. Get stats from DB │
|
|
│ ┌─────────────────────────────────────────────────────┐ │
|
|
│ │ SELECT strategy_id, │ │
|
|
│ │ COUNT(*) as sent, │ │
|
|
│ │ SUM(got_response) as responded │ │
|
|
│ │ FROM sent_messages │ │
|
|
│ │ GROUP BY strategy_id │ │
|
|
│ └─────────────────────────────────────────────────────┘ │
|
|
│ │
|
|
│ 2. Calculate response rates │
|
|
│ variant_a: 15/50 = 30% │
|
|
│ variant_b: 12/30 = 40% │
|
|
│ variant_c: 1/5 = 20% │
|
|
│ │
|
|
│ 3. Filter by minimum samples │
|
|
│ minSamples = 10 │
|
|
│ variant_c excluded (only 5 sends) │
|
|
│ │
|
|
│ 4. Explore vs Exploit │
|
|
│ random() < explorationRate (e.g., 0.15)? │
|
|
│ ├─► Yes: Random pick from enabled variants │
|
|
│ └─► No: Pick highest rate (variant_b at 40%) │
|
|
│ │
|
|
│ Output: selected variant name │
|
|
│ │
|
|
│ ──── UPDATE STATS ────────────────────────────────────────── │
|
|
│ │
|
|
│ Input: strategy_id, got_response (boolean) │
|
|
│ │
|
|
│ 1. Record in sent_messages │
|
|
│ UPDATE sent_messages │
|
|
│ SET got_response = 1, │
|
|
│ response_checked_at = now │
|
|
│ WHERE id = ? │
|
|
│ │
|
|
│ 2. If successful: store in successful_messages │
|
|
│ INSERT INTO successful_messages │
|
|
│ VALUES (original_message, image_analysis, post_title, │
|
|
│ response_snippet, subreddit, now) │
|
|
│ │
|
|
│ 3. Update strategy statistics │
|
|
│ (reflected in next selectStrategy call) │
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
---
|
|
|
|
## Media Sync Flow
|
|
|
|
How media files are downloaded and synchronized from chat conversations.
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ MediaSyncService │
|
|
│ │
|
|
│ ──── TRIGGER ──────────────────────────────────────────────── │
|
|
│ │
|
|
│ Scheduled: every N minutes (config.mediaSync.intervalMinutes) │
|
|
│ Manual: POST /api/media/sync (incremental) │
|
|
│ Manual: POST /api/media/sync/full (full) │
|
|
│ │
|
|
│ ──── INCREMENTAL SYNC ────────────────────────────────────── │
|
|
│ │
|
|
│ 1. Get top 10 conversations from ChatService │
|
|
│ 2. For each conversation: │
|
|
│ ├─► Check if media already downloaded (index lookup) │
|
|
│ ├─► If not: download media │
|
|
│ │ └─► MediaService.downloadMedia(chatId) │
|
|
│ └─► Update index │
|
|
│ │
|
|
│ ──── FULL SYNC ────────────────────────────────────────────── │
|
|
│ │
|
|
│ 1. Get ALL conversations from ChatService │
|
|
│ 2. For each conversation: │
|
|
│ ├─► Download any missing media │
|
|
│ └─► Update index │
|
|
│ │
|
|
│ ──── MEDIA DOWNLOAD (MediaService) ────────────────────────── │
|
|
│ │
|
|
│ 1. Extract media URLs from conversation messages │
|
|
│ 2. For each URL: │
|
|
│ ├─► Check if Matrix thumbnail │
|
|
│ │ └─► Resolve to full resolution URL │
|
|
│ ├─► Download via Playwright │
|
|
│ ├─► Compute SHA-1 hash │
|
|
│ ├─► Check if already exists (dedup) │
|
|
│ ├─► Save to data/media/{hash}.{ext} │
|
|
│ └─► Update index JSON │
|
|
│ │
|
|
│ Output: Array of saved file paths │
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
---
|
|
|
|
## Auto-Reply Flow
|
|
|
|
How the bot automatically replies to chat messages.
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ AutoReplyService │
|
|
│ │
|
|
│ Scheduled: every N minutes (config.autoReply.checkInterval) │
|
|
│ │
|
|
│ 1. Get cached conversations │
|
|
│ 2. For each conversation: │
|
|
│ │ │
|
|
│ ├─► Check for new/unread messages │
|
|
│ │ └─► If no new messages: continue │
|
|
│ │ │
|
|
│ ├─► Detect upsell/spam keywords │
|
|
│ │ ├─► "onlyfans", "link in bio", "check my..." │
|
|
│ │ ├─► If detected: flag conversation, skip reply │
|
|
│ │ └─► If not: continue │
|
|
│ │ │
|
|
│ ├─► Generate AI reply suggestion │
|
|
│ │ ┌─────────────────────────────────────────────────┐ │
|
|
│ │ │ OpenRouter.generateReply(conversationHistory) │ │
|
|
│ │ │ │ │
|
|
│ │ │ Request: { │ │
|
|
│ │ │ messages: [ │ │
|
|
│ │ │ { role: "system", content: replyPrompt }, │ │
|
|
│ │ │ { role: "user", content: conversation } │ │
|
|
│ │ │ ] │ │
|
|
│ │ │ } │ │
|
|
│ │ │ │ │
|
|
│ │ │ Response: "Thanks for reaching out! 😊" │ │
|
|
│ │ │ │ │
|
|
│ │ │ Reply: "Thanks for reaching out! 😊" │ │
|
|
│ │ └─────────────────────────────────────────────────┘ │
|
|
│ │ │
|
|
│ ├─► If autoSend = true: │
|
|
│ │ └─► ChatService.sendMessage(chatId, reply) │
|
|
│ │ └─► Log: "Auto-sent reply to {user}" │
|
|
│ │ │
|
|
│ └─► If autoSend = false: │
|
|
│ └─► Mark as pending (visible in dashboard) │
|
|
│ └─► User can approve/reject via dashboard │
|
|
│ │
|
|
│ Output: Summary of actions taken │
|
|
│ "Checked 15 conversations: 3 new messages, 1 reply sent" │
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
---
|
|
|
|
## Dashboard Data Flow
|
|
|
|
How data flows from services to the React operator console.
|
|
|
|
### Current React model
|
|
|
|
1. Feature-local API clients request authoritative snapshots from `/api/*`.
|
|
2. Shared Zod schemas in `packages/api-contracts` validate HTTP responses and
|
|
realtime envelopes at runtime.
|
|
3. TanStack Query caches snapshots, preserves them during background refresh,
|
|
and owns mutation pending/error/success state.
|
|
4. Operations opens `/api/events/stream`. Each typed event invalidates the
|
|
affected query keys, such as stats, timers, conversations, media, or
|
|
operations activity.
|
|
5. Browser `EventSource` reconnects automatically. The client marks cached
|
|
state stale while reconnecting instead of blanking the UI.
|
|
|
|
```text
|
|
services/repositories -> Express /api snapshots -> Zod -> TanStack Query -> React routes
|
|
| ^
|
|
+-> realtime event hub -> /api/events/stream --------+
|
|
```
|
|
|
|
### Retired legacy model
|
|
|
|
The historical diagram below is retained for migration context only.
|
|
`public/index.html` has been removed and is no longer served.
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ WebServer (Express) │
|
|
│ │
|
|
│ ──── REST API ─────────────────────────────────────────────── │
|
|
│ │
|
|
│ GET /api/stats │
|
|
│ ├─► Read from: │
|
|
│ │ ├─► Bot running state (in-memory) │
|
|
│ │ ├─► PostsRepository.countProcessed() │
|
|
│ │ ├─► MessagesRepository.countSent() │
|
|
│ │ ├─► MessagesRepository.countResponded() │
|
|
│ │ └─► Calculate response rate │
|
|
│ └─► Return: { running, processed, sent, responded, rate } │
|
|
│ │
|
|
│ GET /api/messages │
|
|
│ └─► MessagesRepository.getRecent(limit: 50) │
|
|
│ │
|
|
│ GET /api/analytics/by-strategy │
|
|
│ └─► MessagesRepository.getStrategyStats() │
|
|
│ │
|
|
│ PUT /api/config │
|
|
│ ├─► Validate new config values │
|
|
│ ├─► Update in-memory config │
|
|
│ └─► Return updated config │
|
|
│ │
|
|
│ ──── LEGACY SSE LOG STREAMING ─────────────────────────────── │
|
|
│ │
|
|
│ GET /api/logs/stream │
|
|
│ ├─► Client connects │
|
|
│ ├─► Send any pending log entries │
|
|
│ ├─► Subscribe to logger buffer │
|
|
│ ├─► Push new entries as they arrive │
|
|
│ │ └─► Event: log │
|
|
│ │ └─► Data: { level, message, timestamp } │
|
|
│ └─► On disconnect: unsubscribe │
|
|
│ │
|
|
│ ──── RETIRED LEGACY IMPLEMENTATION ────────────────────────── │
|
|
│ │
|
|
│ JavaScript polls API endpoints: │
|
|
│ ├─► /api/stats every 5 seconds │
|
|
│ ├─► /api/messages every 10 seconds │
|
|
│ ├─► /api/timers every 2 seconds │
|
|
│ └─► SSE /api/logs/stream (persistent connection) │
|
|
│ │
|
|
│ Updates DOM in real-time: │
|
|
│ ├─► Stats cards (processed, sent, responded, rate) │
|
|
│ ├─► Activity feed (scrolling list) │
|
|
│ ├─► Message viewer (click to expand) │
|
|
│ ├─► Chat browser (conversation list) │
|
|
│ ├─► Media gallery (image grid) │
|
|
│ └─► Rate limiter timers (countdown) │
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
---
|
|
|
|
## Error Recovery Flows
|
|
|
|
### AI Refusal Recovery
|
|
|
|
```
|
|
1. OpenRouter.analyzeImage() returns refusal
|
|
│
|
|
├─► Detect refusal via heuristic patterns
|
|
│ ├─► "cannot analyze"
|
|
│ ├─► "against policy"
|
|
│ └─► "unable to process"
|
|
│
|
|
├─► Try next model in fallback chain
|
|
│ ├─► OpenRouter.fallbackModels[0]
|
|
│ ├─► OpenRouter.fallbackModels[1]
|
|
│ └─► ...
|
|
│
|
|
└─► If all models refuse:
|
|
├─► Mark post as processed with error
|
|
├─► Log: "All models refused to analyze post {id}"
|
|
└─► Continue to next post
|
|
```
|
|
|
|
### Network Error Recovery
|
|
|
|
```
|
|
1. RedditApi.fetchPosts() throws error
|
|
│
|
|
└─► withRetry() wrapper handles recovery:
|
|
├─► Retry 1: wait 5s
|
|
├─► Retry 2: wait 15s
|
|
├─► Retry 3: wait 45s
|
|
└─► After 3 failures:
|
|
├─► Log error
|
|
├─► Skip this subreddit
|
|
└─► Continue to next subreddit
|
|
```
|
|
|
|
### Account Session Recovery
|
|
|
|
```
|
|
1. Reddit or Matrix authentication is rejected
|
|
│
|
|
├─► Invalidate the shared account-auth gate
|
|
├─► Stop account-bound schedulers
|
|
└─► Wait for a fresh combined dump from the Chrome extension
|
|
```
|
|
|
|
---
|
|
|
|
## Data Lifecycle
|
|
|
|
### Post Data
|
|
|
|
```
|
|
Discovered (MonitorService)
|
|
│
|
|
├─► Stored in processed_posts (pending)
|
|
│
|
|
├─► Processed (ProcessorService)
|
|
│ ├─► image_analysis updated
|
|
│ ├─► generated_message updated
|
|
│ └─► message_sent updated
|
|
│
|
|
└─► Archived (retention policy TBD)
|
|
└─► Kept indefinitely for analytics
|
|
```
|
|
|
|
### Message Data
|
|
|
|
```
|
|
Generated (ProcessorService)
|
|
│
|
|
├─► Stored in sent_messages (got_response = 0)
|
|
│
|
|
├─► Response checked (ResponseCheckerService)
|
|
│ ├─► got_response = 1
|
|
│ └─► response_checked_at set
|
|
│
|
|
├─► If responded:
|
|
│ └─► Copied to successful_messages (for learning)
|
|
│
|
|
└─► Retained indefinitely for analytics
|
|
```
|
|
|
|
### Learning Data
|
|
|
|
```
|
|
Successful pair created (ResponseCheckerService)
|
|
│
|
|
├─► Stored in successful_messages
|
|
│ ├─► original_message
|
|
│ ├─► image_analysis
|
|
│ ├─► post_title
|
|
│ ├─► response_snippet
|
|
│ └─► subreddit (for weighted selection)
|
|
│
|
|
├─► Retrieved for few-shot learning
|
|
│ └─► Included in OpenRouter.generateMessage() prompt
|
|
│
|
|
└─► Retained indefinitely (grows bot intelligence)
|
|
```
|
|
|
|
### Chat Data
|
|
|
|
```
|
|
Conversation discovered (ChatService)
|
|
│
|
|
├─► Cached in memory
|
|
├─► Persisted to disk JSON
|
|
│
|
|
├─► Messages added as they arrive
|
|
│
|
|
├─► Media downloaded (MediaService)
|
|
│ └─► Stored in data/media/
|
|
│
|
|
└─► Retained until manually cleared
|
|
└─► No automatic cleanup (TBD)
|
|
```
|
|
|
|
### Media Data
|
|
|
|
```
|
|
Media URL discovered
|
|
│
|
|
├─► Downloaded (MediaService)
|
|
│ ├─► Matrix thumbnails resolved
|
|
│ ├─► SHA-1 hash computed
|
|
│ └─► Saved to data/media/{hash}.{ext}
|
|
│
|
|
├─► Indexed in data/media/index.json
|
|
│
|
|
├─► Deduplicated (hash check before download)
|
|
│
|
|
└─► Retained until manually cleared
|
|
└─► No automatic cleanup (TBD)
|
|
```
|