How to link Telegram bot, website and database: bot architecture writes - site reads

─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
versus TELEGRAM
♥ User → writes bot → bot handles ♥
─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ? ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ?
recording
¶
─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
BOT (Separate Process)
♥ Accepts messages, validates, writes in PostgreSQL
─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ? ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ?
versus INSERT/UPDATE
¶
─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
PostgreSQL versus
Posts, users, media are the only source of truth.
─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ? ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ?
SELECT
¶
─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
API-SERVER (Separate Process)
REST API: /api/posts, /api/feed - for the site
─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ? ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ?
versus HTTP/SSE
¶
─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
SITE (frontend) versus
♥ Public feed, filters, real-time updates
─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─

-- Пользователи (из Telegram)
CREATE TABLE users (
    id          SERIAL PRIMARY KEY,
    telegram_id BIGINT UNIQUE NOT NULL,
    username    VARCHAR(64),
    first_name  VARCHAR(128),
    is_banned   BOOLEAN DEFAULT FALSE,
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

-- Посты/сообщения от бота
CREATE TABLE posts (
    id          SERIAL PRIMARY KEY,
    user_id     INTEGER REFERENCES users(id) ON DELETE SET NULL,
    text        TEXT NOT NULL,
    media_url   TEXT,              -- URL медиафайла (если есть)
    media_type  VARCHAR(16),       -- 'photo', 'video', 'document'
    status      VARCHAR(16) DEFAULT 'pending',
    -- pending → approved → published / rejected
    is_public   BOOLEAN DEFAULT FALSE,
    published_at TIMESTAMPTZ,
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

-- Индексы для быстрой публичной ленты
CREATE INDEX idx_posts_public ON posts (published_at DESC)
    WHERE is_public = TRUE AND status = 'published';

CREATE INDEX idx_posts_user ON posts (user_id, created_at DESC);

pending → (moderator approved) → approved → (cron/webhook) → published
) (rejected) → rejected

npm install grammy pg dotenv

// bot.js
const { Bot } = require('grammy');
const { Pool } = require('pg');
require('dotenv').config();

const bot = new Bot(process.env.BOT_TOKEN);
const db = new Pool({ connectionString: process.env.DATABASE_URL });

// Хелпер: найти или создать пользователя
async function upsertUser(telegramUser) {
    const { rows } = await db.query(`
        INSERT INTO users (telegram_id, username, first_name)
        VALUES ($1, $2, $3)
        ON CONFLICT (telegram_id) DO UPDATE
            SET username = EXCLUDED.username,
                first_name = EXCLUDED.first_name
        RETURNING id
    `, [telegramUser.id, telegramUser.username, telegramUser.first_name]);
    return rows[0].id;
}

// Обработка текстового сообщения
bot.on('message:text', async (ctx) => {
    const userId = await upsertUser(ctx.from);

    // Базовая валидация
    const text = ctx.message.text.trim();
    if (text.length < 10) {
        return ctx.reply('Сообщение слишком короткое — минимум 10 символов.');
    }
    if (text.length > 2000) {
        return ctx.reply('Сообщение слишком длинное — максимум 2000 символов.');
    }

    // Записываем в БД
    await db.query(`
        INSERT INTO posts (user_id, text, status, is_public, published_at)
        VALUES ($1, $2, 'published', TRUE, NOW())
    `, [userId, text]);

    await ctx.reply('✅ Опубликовано!');
});

// Обработка фото
bot.on('message:photo', async (ctx) => {
    const userId = await upsertUser(ctx.from);
    const caption = ctx.message.caption?.trim() || '';

    // Берём самое большое фото из массива (Telegram присылает несколько размеров)
    const photo = ctx.message.photo.at(-1);
    const fileId = photo.file_id;

    // Получаем URL файла через Telegram API
    const file = await ctx.api.getFile(fileId);
    const mediaUrl = `https://api.telegram.org/file/bot${process.env.BOT_TOKEN}/${file.file_path}`;

    await db.query(`
        INSERT INTO posts (user_id, text, media_url, media_type, status, is_public, published_at)
        VALUES ($1, $2, $3, 'photo', 'published', TRUE, NOW())
    `, [userId, caption, mediaUrl]);

    await ctx.reply('✅ Фото опубликовано!');
});

// Запуск через вебхук
async function startWebhook() {
    await bot.api.setWebhook(`${process.env.WEBHOOK_URL}/bot`);
    console.log('Вебхук установлен');
}

module.exports = { bot, startWebhook };

// server.js
const express = require('express');
const { webhookCallback } = require('grammy');
const { bot, startWebhook } = require('./bot');

const app = express();
app.use(express.json());

// Эндпоинт для вебхука Telegram
app.post('/bot', webhookCallback(bot, 'express'));

app.listen(3000, async () => {
    console.log('Сервер запущен на порту 3000');
    await startWebhook();
});

// api.js — отдельный файл/процесс
const express = require('express');
const { Pool } = require('pg');

const app = express();
const db = new Pool({ connectionString: process.env.DATABASE_URL });

// CORS для фронтенда
app.use((req, res, next) => {
    res.setHeader('Access-Control-Allow-Origin', process.env.FRONTEND_URL || '*');
    res.setHeader('Access-Control-Allow-Methods', 'GET');
    next();
});

// Публичная лента: последние посты
app.get('/api/feed', async (req, res) => {
    const page = parseInt(req.query.page) || 1;
    const limit = Math.min(parseInt(req.query.limit) || 20, 100);
    const offset = (page - 1) * limit;

    const { rows } = await db.query(`
        SELECT
            p.id,
            p.text,
            p.media_url,
            p.media_type,
            p.published_at,
            u.username,
            u.first_name
        FROM posts p
        LEFT JOIN users u ON p.user_id = u.id
        WHERE p.is_public = TRUE
          AND p.status = 'published'
          AND (u.id IS NULL OR u.is_banned = FALSE)
        ORDER BY p.published_at DESC
        LIMIT $1 OFFSET $2
    `, [limit, offset]);

    // Общее количество для пагинации
    const { rows: countRows } = await db.query(`
        SELECT COUNT(*) FROM posts
        WHERE is_public = TRUE AND status = 'published'
    `);

    res.json({
        items: rows,
        total: parseInt(countRows[0].count),
        page,
        limit,
    });
});

// Один пост по ID
app.get('/api/posts/:id', async (req, res) => {
    const { rows } = await db.query(`
        SELECT p.*, u.username, u.first_name
        FROM posts p
        LEFT JOIN users u ON p.user_id = u.id
        WHERE p.id = $1 AND p.is_public = TRUE AND p.status = 'published'
    `, [req.params.id]);

    if (!rows.length) return res.status(404).json({ error: 'Не найдено' });
    res.json(rows[0]);
});

app.listen(4000, () => console.log('API на порту 4000'));

// frontend.js
let lastId = 0;

async function pollNewPosts() {
    const res = await fetch(`/api/feed?since_id=${lastId}`);
    const { items } = await res.json();

    if (items.length > 0) {
        lastId = items[0].id;
        items.reverse().forEach(post => prependPost(post));
    }
}

// Опрашиваем каждые 10 секунд
setInterval(pollNewPosts, 10_000);

// В API добавляем параметр since_id
app.get('/api/feed', async (req, res) => {
    const sinceId = parseInt(req.query.since_id) || 0;
    // ...
    const { rows } = await db.query(`
        SELECT ... FROM posts p
        WHERE p.is_public = TRUE
          AND p.status = 'published'
          AND ($1 = 0 OR p.id > $1)  -- только новее last_id
        ORDER BY p.published_at DESC
        LIMIT $2 OFFSET $3
    `, [sinceId, limit, offset]);
});

// SSE-эндпоинт в API
const clients = new Set();

app.get('/api/feed/stream', (req, res) => {
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Connection', 'keep-alive');

    // Регистрируем клиента
    clients.add(res);

    // Убираем при отключении
    req.on('close', () => clients.delete(res));

    // Heartbeat — чтобы соединение не рвалось
    const heartbeat = setInterval(() => {
        res.write(': ping\n\n');
    }, 30_000);

    req.on('close', () => clearInterval(heartbeat));
});

// Эту функцию вызываем из бота после успешной записи в БД
function notifyNewPost(post) {
    const data = JSON.stringify(post);
    clients.forEach(client => {
        client.write(`event: new_post\ndata: ${data}\n\n`);
    });
}

module.exports = { notifyNewPost };

// bot.js — уведомляем SSE после записи
const { notifyNewPost } = require('./api');

// ... после INSERT:
const { rows } = await db.query(`
    INSERT INTO posts (...) VALUES (...) RETURNING *
`);
notifyNewPost(rows[0]);

// frontend.js — подписка на SSE
const evtSource = new EventSource('/api/feed/stream');

evtSource.addEventListener('new_post', (event) => {
    const post = JSON.parse(event.data);
    prependPost(post); // добавляем в начало ленты
});

evtSource.onerror = () => {
    // Браузер автоматически переподключается при разрыве
    console.log('SSE переподключение...');
};

// Бот публикует событие в Redis после записи
const redis = require('redis');
const publisher = redis.createClient({ url: process.env.REDIS_URL });

async function publishNewPost(post) {
    await publisher.publish('new_post', JSON.stringify(post));
}

// API подписывается на Redis и рассылает через SSE
const subscriber = redis.createClient({ url: process.env.REDIS_URL });

await subscriber.subscribe('new_post', (message) => {
    const post = JSON.parse(message);
    clients.forEach(client => {
        client.write(`event: new_post\ndata: ${message}\n\n`);
    });
});

const redis = require('redis');
const cache = redis.createClient({ url: process.env.REDIS_URL });

const CACHE_TTL = 30; // секунд

app.get('/api/feed', async (req, res) => {
    const cacheKey = `feed:page:${req.query.page || 1}`;

    // Проверяем кэш
    const cached = await cache.get(cacheKey);
    if (cached) {
        res.setHeader('X-Cache', 'HIT');
        return res.json(JSON.parse(cached));
    }

    // Идём в БД
    const data = await fetchFeedFromDB(req.query);

    // Кладём в кэш
    await cache.setEx(cacheKey, CACHE_TTL, JSON.stringify(data));

    res.setHeader('X-Cache', 'MISS');
    res.json(data);
});

// После записи нового поста — инвалидируем кэш
async function invalidateFeedCache() {
    const keys = await cache.keys('feed:*');
    if (keys.length) await cache.del(keys);
}

// ecosystem.config.js
module.exports = {
    apps: [
        {
            name: 'bot',
            script: './bot/server.js',
            env: { NODE_ENV: 'production', PORT: 3000 },
        },
        {
            name: 'api',
            script: './api/server.js',
            env: { NODE_ENV: 'production', PORT: 4000 },
        },
    ],
};

pm2 start ecosystem.config. js
pm2 save
pm2 startup # autoboot when restarting

# docker-compose.yml
version: '3.9'

services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: myapp
      POSTGRES_USER: myapp
      POSTGRES_PASSWORD: ${DB_PASSWORD}
    volumes:
      - pgdata:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    restart: unless-stopped

  bot:
    build: ./bot
    environment:
      DATABASE_URL: postgresql://myapp:${DB_PASSWORD}@postgres/myapp
      REDIS_URL: redis://redis:6379
      BOT_TOKEN: ${BOT_TOKEN}
      WEBHOOK_URL: ${WEBHOOK_URL}
    depends_on:
      - postgres
      - redis
    restart: unless-stopped

  api:
    build: ./api
    ports:
      - "4000:4000"
    environment:
      DATABASE_URL: postgresql://myapp:${DB_PASSWORD}@postgres/myapp
      REDIS_URL: redis://redis:6379
    depends_on:
      - postgres
      - redis
    restart: unless-stopped

volumes:
  pgdata:

server {
    server_name your-site.ru;

    # Статика фронтенда
    location / {
        root /var/www/frontend/dist;
        try_files $uri $uri/ /index.html;
    }

    # API
    location /api/ {
        proxy_pass http://localhost:4000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }

    # SSE — важно: отключить буферизацию
    location /api/feed/stream {
        proxy_pass http://localhost:4000;
        proxy_buffering off;
        proxy_cache off;
        proxy_set_header Connection '';
        proxy_http_version 1.1;
        chunked_transfer_encoding on;
    }

    # Вебхук для бота
    location /bot {
        proxy_pass http://localhost:3000;
    }
}

-- Пользователь для бота: пишет
CREATE USER bot_user WITH PASSWORD 'strong_password_1';
GRANT CONNECT ON DATABASE myapp TO bot_user;
GRANT USAGE ON SCHEMA public TO bot_user;
GRANT INSERT, UPDATE ON posts TO bot_user;
GRANT INSERT, UPDATE ON users TO bot_user;
GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO bot_user;

-- Пользователь для API: только читает
CREATE USER api_user WITH PASSWORD 'strong_password_2';
GRANT CONNECT ON DATABASE myapp TO api_user;
GRANT USAGE ON SCHEMA public TO api_user;
GRANT SELECT ON posts, users TO api_user;

const { S3Client, PutObjectCommand } = require('@aws-sdk/client-s3');
const https = require('https');

const s3 = new S3Client({
    endpoint: process.env.S3_ENDPOINT,
    credentials: {
        accessKeyId: process.env.S3_KEY,
        secretAccessKey: process.env.S3_SECRET,
    },
    region: 'ru-1',
});

async function uploadTelegramPhoto(fileId, botToken) {
    // Получаем URL файла
    const file = await bot.api.getFile(fileId);
    const tgUrl = `https://api.telegram.org/file/bot${botToken}/${file.file_path}`;
    const ext = file.file_path.split('.').pop();

    // Скачиваем
    const buffer = await new Promise((resolve, reject) => {
        https.get(tgUrl, (res) => {
            const chunks = [];
            res.on('data', c => chunks.push(c));
            res.on('end', () => resolve(Buffer.concat(chunks)));
            res.on('error', reject);
        });
    });

    // Загружаем в S3
    const key = `uploads/${Date.now()}.${ext}`;
    await s3.send(new PutObjectCommand({
        Bucket: process.env.S3_BUCKET,
        Key: key,
        Body: buffer,
        ContentType: `image/${ext}`,
        ACL: 'public-read',
    }));

    return `${process.env.S3_PUBLIC_URL}/${key}`;
}

const MODERATOR_CHAT_ID = process.env.MODERATOR_CHAT_ID;

// Новый пост — сначала pending
bot.on('message:text', async (ctx) => {
    const userId = await upsertUser(ctx.from);

    const { rows } = await db.query(`
        INSERT INTO posts (user_id, text, status, is_public)
        VALUES ($1, $2, 'pending', FALSE)
        RETURNING id
    `, [userId, ctx.message.text]);

    const postId = rows[0].id;

    // Шлём модератору
    await bot.api.sendMessage(MODERATOR_CHAT_ID,
        `📨 Новый пост #${postId} от @${ctx.from.username}:\n\n${ctx.message.text}`,
        {
            reply_markup: {
                inline_keyboard: [[
                    { text: '✅ Одобрить', callback_data: `approve:${postId}` },
                    { text: '❌ Отклонить', callback_data: `reject:${postId}` },
                ]],
            },
        }
    );

    await ctx.reply('Пост отправлен на модерацию.');
});

// Модератор нажимает кнопку
bot.on('callback_query:data', async (ctx) => {
    const [action, postId] = ctx.callbackQuery.data.split(':');

    if (action === 'approve') {
        await db.query(`
            UPDATE posts
            SET status = 'published', is_public = TRUE, published_at = NOW()
            WHERE id = $1
        `, [postId]);
        await ctx.editMessageText(`✅ Пост #${postId} опубликован.`);
    }

    if (action === 'reject') {
        await db.query(`
            UPDATE posts SET status = 'rejected' WHERE id = $1
        `, [postId]);
        await ctx.editMessageText(`❌ Пост #${postId} отклонён.`);
    }

    await ctx.answerCallbackQuery();
});

my-project
─ bot/
← Server.js # Express + Webhook
←  ─ bot.js # Logic Bot
♥ ─ db.js #Pool connections
←  ─ package.json
─ api/
← Server.js # Express API
♥  ─ routes/
♥  ─ feed.js # /api/feed, /api/posts/:id
←  ─ Middleware/
♥  ─ cache.js #redis-cache
←  ─ package.json
─ frontend/
←  ─ index.html
♥ ы─ feed.js # Ribbon download, SSE
♥  ─ style.css
а ─ init.sql # BD Scheme
─ docker-compose.yml
.─ Ecosystem.config. js
─ .env

BG:
Users and posts tables are created with the necessary indexes
Index on published at WHERE is public = TRUE
Two DB users: bot user (INSERT) and api user (SELECT)
● Status field for float pending → published / rejected

Bot:
● Works through webhook, not polling
Validation of length and content to INSERT
Media files are uploaded to S3, not stored as file id
● After recording: Cache invalidation and SSE/Redis notification

API:
SELECT only, no INSERT/UPDATE in the public API
● Pagination at all endpoints
Redis cache on disabled tape
CORS is configured for a specific domain, not *

Realtime:
Selected method: polling / SSE / Redis Pub / Sub
For SSE: Nginx with buffering disabled

Deploy:
● Bot and API – different processes (PM2 or Docker)
Nginx as reverse proxy
HTTPS via Let's Encrypt
● Environment variables in .env, not in code