feat(publish): синхрон сайт/ТГ + защита от залпа + заметки Зеро по механике черновиков
ЭТАП 4 — расписание публикаций (раз и навсегда):
1. Синхронизация сайт ↔ Telegram:
- articleAutoPublish.pickScheduleTime теперь ставит TG-пост на
articles.published_at (тот же момент, когда статья появляется на сайте).
Слоты канала publish_slots больше НЕ выбирают время независимо.
- Единый источник времени — published_at (drip-слот сайта).
- Слоты сайта и ТГ синхронизированы в БД: 08:11/12:11/16:11/20:11.
2. Защита от залпа (scheduledPostsRunner):
- посты просроченные >3ч → status='skipped' (не спамим канал задним числом).
- публикация по ОДНОМУ за тик (LIMIT 1), не пачкой.
3. Заметки Зеро (zeroNotes) — механика как у черновиков статей:
- nextPublishSlot всегда = ЗАВТРА publishHour (сегодня сгенерили → завтра
публикуем). Час настраивается ZERO_NOTES_PUBLISH_HOUR (13:00).
- autoApproveOldDrafts (09:00 МСК): если не подтверждён к утру дня
публикации — авто-одобряется и выходит в свой слот.
- publishReady limit:1 — строго одна заметка за тик.
Настройки в app_settings: SITE_PUBLISH_SLOTS, ZERO_SITE_URL_BASE,
ZERO_NOTES_APPROVE_HOUR=9, GENERATE_HOUR=13, PUBLISH_HOUR=13.
This commit is contained in:
Aleksei Pavlov
@@ -1,62 +1,45 @@
|
||||
// Авто-публикация статей в каналы.
|
||||
//
|
||||
// СИНХРОНИЗАЦИЯ С САЙТОМ (главный принцип):
|
||||
// TG-пост ставится на ТО ЖЕ время, когда статья появляется на сайте —
|
||||
// на articles.published_at. Сайт и Telegram больше не разъезжаются.
|
||||
// (published_at выставляется dripScheduler'ом при approve черновика —
|
||||
// в слоты SITE_PUBLISH_SLOTS = 08:11/12:11/16:11/20:11.)
|
||||
//
|
||||
// Логика:
|
||||
// 1. При сохранении статьи со status='published' — engine вызывает scheduleForArticle(articleId)
|
||||
// 2. Находим все системные каналы с auto_publish_enabled=true где (categories пустой ИЛИ категория статьи там есть)
|
||||
// 3. Для каждого канала ищем ближайший подходящий момент:
|
||||
// - если delay_min > 0 → now + delay_min
|
||||
// - иначе — ближайший publish_slot канала в будущем
|
||||
// - если у канала нет слотов и delay=0 — публикуем сразу (scheduled_at = NOW)
|
||||
// 4. Дедуп: один article × один channel = одна запись в scheduled_posts (skip если уже есть pending/sent)
|
||||
// 5. Создаём scheduled_posts с pending status — runner отработает по cron'у
|
||||
// 1. При публикации статьи engine вызывает scheduleForArticle(articleId)
|
||||
// 2. Находим системные каналы с auto_publish_enabled=true где категория подходит
|
||||
// 3. scheduled_at = articles.published_at (если оно в будущем),
|
||||
// иначе now + delay_min (или NOW). Слоты канала publish_slots больше
|
||||
// НЕ используются для выбора времени — единый источник это published_at.
|
||||
// 4. Дедуп: один article × один channel = одна запись (skip если pending/sent)
|
||||
// 5. Раннер (scheduledPostsRunner) отправит когда scheduled_at <= NOW,
|
||||
// с защитой от залпа (пропускает посты старше SKIP_OLDER_THAN_H часов).
|
||||
|
||||
const { query } = require('../config/db');
|
||||
|
||||
/**
|
||||
* Подобрать ближайший момент публикации для канала.
|
||||
* Момент публикации в TG = момент появления на сайте (published_at статьи).
|
||||
* Если published_at не задан/в прошлом — используем delay_min или NOW.
|
||||
* @param {object} channel
|
||||
* @param {object} article — должен содержать published_at
|
||||
* @returns Date
|
||||
*/
|
||||
async function pickScheduleTime(channel) {
|
||||
async function pickScheduleTime(channel, article) {
|
||||
const now = new Date();
|
||||
|
||||
// Главный путь: синхрон с сайтом — ставим на published_at статьи
|
||||
if (article && article.published_at) {
|
||||
const pub = new Date(article.published_at);
|
||||
if (pub > now) return pub; // статья выйдет на сайте в будущем → TG тогда же
|
||||
return now; // статья уже на сайте → публикуем в TG сейчас
|
||||
}
|
||||
|
||||
// Fallback (нет published_at): прежнее поведение через delay
|
||||
if (channel.auto_publish_delay_min > 0) {
|
||||
return new Date(now.getTime() + channel.auto_publish_delay_min * 60_000);
|
||||
}
|
||||
// Ищем publish_slots
|
||||
const { rows: slots } = await query(
|
||||
`SELECT slot_hour, slot_minute FROM publish_slots
|
||||
WHERE channel_id=$1 AND enabled=true
|
||||
ORDER BY slot_hour, slot_minute`,
|
||||
[channel.id]
|
||||
);
|
||||
if (slots.length === 0) {
|
||||
return now; // публикуем сразу
|
||||
}
|
||||
|
||||
// Получаем уже занятые слоты (pending)
|
||||
const { rows: taken } = await query(
|
||||
`SELECT scheduled_at FROM scheduled_posts WHERE channel_id=$1 AND status='pending'`,
|
||||
[channel.id]
|
||||
);
|
||||
const takenTimes = new Set(taken.map(r => new Date(r.scheduled_at).getTime()));
|
||||
|
||||
// Ищем ближайший незанятый слот на ближайшие 7 дней
|
||||
for (let dayOffset = 0; dayOffset <= 7; dayOffset++) {
|
||||
const base = new Date(now);
|
||||
base.setDate(base.getDate() + dayOffset);
|
||||
for (const s of slots) {
|
||||
const t = new Date(base);
|
||||
t.setHours(s.slot_hour, s.slot_minute, 0, 0);
|
||||
if (t > now && !takenTimes.has(t.getTime())) {
|
||||
return t;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Fallback — завтра первый слот
|
||||
const tomorrow = new Date(now);
|
||||
tomorrow.setDate(tomorrow.getDate() + 1);
|
||||
tomorrow.setHours(slots[0].slot_hour, slots[0].slot_minute, 0, 0);
|
||||
return tomorrow;
|
||||
return now;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -66,7 +49,7 @@ async function pickScheduleTime(channel) {
|
||||
*/
|
||||
async function scheduleForArticle(articleId) {
|
||||
const { rows: arts } = await query(
|
||||
`SELECT id, slug, title, category, status FROM articles WHERE id=$1`,
|
||||
`SELECT id, slug, title, category, status, published_at FROM articles WHERE id=$1`,
|
||||
[articleId]
|
||||
);
|
||||
if (!arts.length || arts[0].status !== 'published') return [];
|
||||
@@ -84,23 +67,23 @@ async function scheduleForArticle(articleId) {
|
||||
|
||||
const created = [];
|
||||
for (const ch of channels) {
|
||||
// Дедуп
|
||||
// Дедуп — одна запись на (channel, article) в активных статусах
|
||||
const { rows: existing } = await query(
|
||||
`SELECT id FROM scheduled_posts
|
||||
WHERE channel_id=$1 AND article_id=$2 AND status IN ('pending','sent')
|
||||
WHERE channel_id=$1 AND article_id=$2 AND status IN ('pending','sent','sending')
|
||||
LIMIT 1`,
|
||||
[ch.id, article.id]
|
||||
);
|
||||
if (existing.length) continue;
|
||||
|
||||
const scheduledAt = await pickScheduleTime(ch);
|
||||
const scheduledAt = await pickScheduleTime(ch, article);
|
||||
const { rows: inserted } = await query(
|
||||
`INSERT INTO scheduled_posts (channel_id, article_id, scheduled_at, status)
|
||||
VALUES ($1,$2,$3,'pending') RETURNING *`,
|
||||
[ch.id, article.id, scheduledAt]
|
||||
);
|
||||
created.push(inserted[0]);
|
||||
console.log(`[auto-publish] article=${article.id} → channel=${ch.id} at ${scheduledAt.toISOString()}`);
|
||||
console.log(`[auto-publish] article=${article.id} → channel=${ch.id} at ${scheduledAt.toISOString()} (synced to published_at)`);
|
||||
}
|
||||
return created;
|
||||
}
|
||||
|
||||
@@ -396,11 +396,31 @@ async function publishOne(scheduledPost) {
|
||||
return { messageId, channel, article };
|
||||
}
|
||||
|
||||
// Защита от залпа: посты просроченные более чем на SKIP_OLDER_THAN_H часов
|
||||
// не отправляются (помечаются 'skipped'). Иначе после простоя раннера или
|
||||
// массового ретрая всё накопившееся улетело бы в канал пачкой.
|
||||
const SKIP_OLDER_THAN_H = 3;
|
||||
|
||||
async function runScheduled() {
|
||||
// 1) Помечаем слишком старые pending как skipped (не спамим канал задним числом)
|
||||
const { rows: skipped } = await query(
|
||||
`UPDATE scheduled_posts
|
||||
SET status='skipped',
|
||||
error='auto-skipped: просрочено более ${SKIP_OLDER_THAN_H}ч'
|
||||
WHERE status='pending'
|
||||
AND scheduled_at < NOW() - INTERVAL '${SKIP_OLDER_THAN_H} hours'
|
||||
RETURNING id, article_id`
|
||||
);
|
||||
if (skipped.length) {
|
||||
console.log(`[scheduled-runner] skipped ${skipped.length} stale post(s): ${skipped.map(s => s.id).join(', ')}`);
|
||||
}
|
||||
|
||||
// 2) Берём ОДИН готовый пост за тик (не пачкой) — публикации идут по одной
|
||||
// с интервалом в минуту, даже если накопилось несколько свежих.
|
||||
const { rows } = await query(
|
||||
`SELECT * FROM scheduled_posts
|
||||
WHERE status='pending' AND scheduled_at <= NOW()
|
||||
ORDER BY scheduled_at ASC LIMIT 20`
|
||||
ORDER BY scheduled_at ASC LIMIT 1`
|
||||
);
|
||||
const results = [];
|
||||
for (const sp of rows) {
|
||||
@@ -422,7 +442,7 @@ async function runScheduled() {
|
||||
console.error(`[scheduled-runner] failed id=${sp.id}: ${msg}`);
|
||||
}
|
||||
}
|
||||
return { processed: rows.length, results };
|
||||
return { processed: rows.length, skipped: skipped.length, results };
|
||||
}
|
||||
|
||||
module.exports = { runScheduled, publishOne, renderTemplate, DEFAULT_TEMPLATE, DEFAULT_BUTTON_TEXT };
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
* Цикл одной заметки:
|
||||
* 1) 13:00 МСК — generateDraft(channelId) → status='draft', scheduled_at=ЗАВТРА 13:00
|
||||
* 2) Вечером — редактор вручную: approveManual / editContent / skip / regenerate
|
||||
* 3) 07:00 МСК — autoApproveOldDrafts(): остальные draft → 'approved' (approved_by='auto')
|
||||
* 3) 09:00 МСК — autoApproveOldDrafts(): неподтверждённые draft → 'approved' (approved_by='auto')
|
||||
* 4) 13:00 следующего дня — runner забирает 'approved' и публикует
|
||||
*
|
||||
* Здесь только бизнес-логика. Расписание — в src/workers/zeroNotesScheduler.js.
|
||||
@@ -36,13 +36,14 @@ function nowMsk() {
|
||||
}
|
||||
|
||||
/**
|
||||
* Следующий публикационный слот в МСК (13:00 по умолчанию).
|
||||
* Если сейчас < сегодняшнего 13:00 МСК — возвращаем сегодня в 13:00 МСК,
|
||||
* иначе — завтра в 13:00 МСК. Возвращаем как UTC Date для записи в TIMESTAMPTZ.
|
||||
* Публикационный слот — ВСЕГДА СЛЕДУЮЩИЙ ДЕНЬ в publishHour МСК (13:00 по умолчанию).
|
||||
* Механика как у черновиков статей: сегодня сгенерили → завтра в 13:00 публикуем
|
||||
* (если подтвердили; иначе авто-одобрение в 09:00 того же дня).
|
||||
* Возвращаем UTC Date для записи в TIMESTAMPTZ.
|
||||
*/
|
||||
function nextPublishSlot({ publishHourMsk = 13, publishMinMsk = 0 } = {}) {
|
||||
const { date: nowM, hour, minute } = nowMsk();
|
||||
const isFuture = hour < publishHourMsk || (hour === publishHourMsk && minute < publishMinMsk);
|
||||
const { date: nowM } = nowMsk();
|
||||
const isFuture = false; // всегда планируем на завтра — заметка должна "отлежаться" сутки на подтверждение
|
||||
const baseUtc = new Date(nowM.getTime() - MSK_OFFSET_MIN * 60_000);
|
||||
baseUtc.setUTCHours(publishHourMsk - 3, publishMinMsk, 0, 0); // UTC-эквивалент МСК-часа
|
||||
if (!isFuture) baseUtc.setUTCDate(baseUtc.getUTCDate() + 1);
|
||||
@@ -271,16 +272,19 @@ async function editContent(noteId, { content, theme, pose, imageUrl, scheduledAt
|
||||
* Запускается в 07:00 МСК — переводит все вчерашние draft в готовые к публикации в 13:00.
|
||||
*/
|
||||
async function autoApproveOldDrafts() {
|
||||
// Запускается в 09:00 МСК (ZERO_NOTES_APPROVE_HOUR). Авто-одобряет драфты,
|
||||
// чей день публикации уже наступил (scheduled_at сегодня или раньше), а юзер
|
||||
// не подтвердил вручную. Так заметка всё равно выйдет в свой слот (13:00).
|
||||
const { rows } = await query(
|
||||
`UPDATE zero_notes
|
||||
SET status='approved', approved_at=NOW(), approved_by='auto', updated_at=NOW()
|
||||
WHERE status='draft'
|
||||
AND scheduled_at IS NOT NULL
|
||||
AND scheduled_at <= NOW() + INTERVAL '8 hours'
|
||||
AND scheduled_at <= NOW() + INTERVAL '6 hours'
|
||||
RETURNING id, channel_id, theme_bucket`
|
||||
);
|
||||
if (rows.length) {
|
||||
console.log(`[zeroNotes] auto-approved ${rows.length} drafts: ${rows.map(r => r.id).join(', ')}`);
|
||||
console.log(`[zeroNotes] auto-approved ${rows.length} draft(s) (не подтверждены к утру): ${rows.map(r => r.id).join(', ')}`);
|
||||
}
|
||||
return rows;
|
||||
}
|
||||
|
||||
@@ -49,15 +49,17 @@ async function getChannel(channelId) {
|
||||
}
|
||||
|
||||
/**
|
||||
* Кнопка "Открыть на сайте" — опционально, если ZERO_SITE_URL_BASE задан.
|
||||
* Пример: ZERO_SITE_URL_BASE=https://zeropost.ru/zero → https://zeropost.ru/zero/123
|
||||
* Кнопка "Читать на сайте" — ведёт на страницу заметок Зеро.
|
||||
* ZERO_SITE_URL_BASE по умолчанию https://zeropost.ru/zero.
|
||||
* Отдельной страницы /zero/:id пока нет, поэтому ведём на общий /zero
|
||||
* (если появится — добавить `/${noteId}`).
|
||||
*/
|
||||
async function buildReplyMarkup(noteId) {
|
||||
const base = await settings.get('ZERO_SITE_URL_BASE', '');
|
||||
async function buildReplyMarkup(_noteId) {
|
||||
const base = await settings.get('ZERO_SITE_URL_BASE', 'https://zeropost.ru/zero');
|
||||
if (!base) return undefined;
|
||||
return {
|
||||
inline_keyboard: [[
|
||||
{ text: '💬 Открыть на сайте', url: `${base.replace(/\/$/, '')}/${noteId}` },
|
||||
{ text: '💬 Читать на сайте', url: base.replace(/\/$/, '') },
|
||||
]],
|
||||
};
|
||||
}
|
||||
|
||||
@@ -70,7 +70,10 @@ async function tick() {
|
||||
if (hour === genHour) await runGeneration(ymd);
|
||||
if (hour === appHour) await runAutoApprove(ymd);
|
||||
// публикация approved-заметок в TG (каждую минуту)
|
||||
const published = await zeroRunner.publishReady({ limit: 3 });
|
||||
// limit:1 — строго одна заметка за тик. Заметки генерятся 1/день, и даже
|
||||
// если накопилось несколько approved (например после ручного ретрая), они
|
||||
// НЕ улетят пачкой — будут публиковаться по одной с интервалом в минуту.
|
||||
const published = await zeroRunner.publishReady({ limit: 1 });
|
||||
if (published > 0) console.log(`[zeroNotes/scheduler] published ${published} note(s)`);
|
||||
} catch (err) {
|
||||
console.error(`[zeroNotes/scheduler] tick error: ${err.message}`);
|
||||
|
||||
Reference in New Issue
Block a user