feat: user_posts service — draft/scheduled/published, Telegram publish with image, cron-driven scheduled publication

src/services/userPosts.js

@@ -0,0 +1,145 @@
+const { query } = require('../config/db');
+const axios = require('axios');
+
+/**
+ * Сохранить пост в базу (как черновик или сразу запланированный).
+ */
+async function savePost({ userId, channelId, content, imageUrl = null, topic = null, status = 'draft', scheduledAt = null }) {
+  const { rows } = await query(
+    `INSERT INTO user_posts (user_id, channel_id, content, image_url, topic, status, scheduled_at)
+     VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING *`,
+    [userId, channelId, content, imageUrl, topic, status, scheduledAt]
+  );
+  return rows[0];
+}
+
+async function listUserPosts({ userId, channelId = null, status = null, limit = 50 }) {
+  let sql = `SELECT * FROM user_posts WHERE user_id=$1`;
+  const params = [userId];
+  if (channelId) { sql += ` AND channel_id=$${params.length + 1}`; params.push(channelId); }
+  if (status) { sql += ` AND status=$${params.length + 1}`; params.push(status); }
+  sql += ` ORDER BY created_at DESC LIMIT $${params.length + 1}`;
+  params.push(limit);
+  const { rows } = await query(sql, params);
+  return rows;
+}
+
+async function getPost(userId, postId) {
+  const { rows } = await query(`SELECT * FROM user_posts WHERE id=$1 AND user_id=$2`, [postId, userId]);
+  return rows[0] || null;
+}
+
+async function updatePost(userId, postId, data) {
+  const allowed = ['content','image_url','status','scheduled_at','topic'];
+  const fields = []; const vals = []; let i = 1;
+  for (const key of allowed) {
+    if (data[key] !== undefined) { fields.push(`${key}=$${i++}`); vals.push(data[key]); }
+  }
+  if (!fields.length) return null;
+  fields.push(`updated_at=NOW()`);
+  vals.push(postId, userId);
+  const { rows } = await query(
+    `UPDATE user_posts SET ${fields.join(',')} WHERE id=$${i++} AND user_id=$${i} RETURNING *`,
+    vals
+  );
+  return rows[0];
+}
+
+async function deletePost(userId, postId) {
+  const { rowCount } = await query(`DELETE FROM user_posts WHERE id=$1 AND user_id=$2`, [postId, userId]);
+  return rowCount > 0;
+}
+
+/**
+ * Опубликовать пост в Telegram.
+ */
+async function publishToTelegram(post, channel) {
+  if (!channel.bot_token || !channel.tg_channel_id) {
+    throw new Error('Telegram не настроен (нужны bot_token и tg_channel_id)');
+  }
+  // Если есть картинка — отправляем как фото с подписью, иначе текст
+  if (post.image_url) {
+    const photoUrl = post.image_url.startsWith('http')
+      ? post.image_url
+      : `https://app.zeropost.ru${post.image_url}`;
+    const res = await axios.post(
+      `https://api.telegram.org/bot${channel.bot_token}/sendPhoto`,
+      {
+        chat_id: channel.tg_channel_id,
+        photo: photoUrl,
+        caption: post.content.slice(0, 1024),
+        parse_mode: 'Markdown',
+      },
+      { timeout: 30000 }
+    );
+    return { ok: true, message_id: res.data?.result?.message_id };
+  } else {
+    const res = await axios.post(
+      `https://api.telegram.org/bot${channel.bot_token}/sendMessage`,
+      {
+        chat_id: channel.tg_channel_id,
+        text: post.content,
+        parse_mode: 'Markdown',
+        disable_web_page_preview: false,
+      },
+      { timeout: 15000 }
+    );
+    return { ok: true, message_id: res.data?.result?.message_id };
+  }
+}
+
+/**
+ * Опубликовать пост (выбирает платформу по channel.platform).
+ */
+async function publishPost(userId, postId) {
+  const post = await getPost(userId, postId);
+  if (!post) throw new Error('Post not found');
+
+  const { rows: chRows } = await query(`SELECT * FROM channels WHERE id=$1`, [post.channel_id]);
+  if (!chRows.length) throw new Error('Channel not found');
+  const channel = chRows[0];
+
+  let result;
+  try {
+    if (channel.platform === 'telegram' || !channel.platform) {
+      result = await publishToTelegram(post, channel);
+    } else {
+      throw new Error(`Платформа ${channel.platform} пока не поддерживается`);
+    }
+    await query(
+      `UPDATE user_posts SET status='published', published_at=NOW(), tg_message_id=$1, error=NULL WHERE id=$2`,
+      [result.message_id || null, postId]
+    );
+    return { ok: true, message_id: result.message_id };
+  } catch (err) {
+    const msg = err.response?.data?.description || err.message;
+    await query(
+      `UPDATE user_posts SET status='failed', error=$1 WHERE id=$2`,
+      [msg, postId]
+    );
+    throw new Error(msg);
+  }
+}
+
+/**
+ * Запустить публикацию запланированных постов (вызывается cron-ом).
+ */
+async function runScheduledPublications() {
+  const { rows } = await query(
+    `SELECT * FROM user_posts
+     WHERE status='scheduled' AND scheduled_at <= NOW()
+     ORDER BY scheduled_at LIMIT 50`
+  );
+  const results = [];
+  for (const post of rows) {
+    try {
+      const res = await publishPost(post.user_id, post.id);
+      results.push({ id: post.id, ok: true, message_id: res.message_id });
+    } catch (err) {
+      results.push({ id: post.id, ok: false, error: err.message });
+    }
+  }
+  return { processed: rows.length, results };
+}
+
+module.exports = { savePost, listUserPosts, getPost, updatePost, deletePost, publishPost, runScheduledPublications };