mirror of
https://github.com/R0m1k3/Socialflow.git
synced 2026-10-23 06:58:12 +02:00
Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
57372c04e9 | ||
|
|
719e57c04b | ||
|
|
def7d8a2cb | ||
|
|
7f21e23a91 |
No files matched your search
@@ -1,6 +1,6 @@
|
||||
import { useQuery } from "@tanstack/react-query";
|
||||
import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "@/components/ui/card";
|
||||
import { ArrowRight, Clock, CheckCircle2, XCircle, Eye, Image as ImageIcon, Smartphone, Clapperboard } from "lucide-react";
|
||||
import { ArrowRight, Clock, CheckCircle2, XCircle, RotateCw, Eye, Image as ImageIcon, Smartphone, Clapperboard } from "lucide-react";
|
||||
import { Link } from "wouter";
|
||||
import { Badge } from "@/components/ui/badge";
|
||||
import { EmptyState } from "@/components/empty-state";
|
||||
@@ -197,7 +197,11 @@ export default function RecentPublications() {
|
||||
<p className="truncate text-sm font-medium" data-testid={`text-page-name-${scheduledPost.id}`}>
|
||||
{scheduledPost.page?.pageName || "Page inconnue"}
|
||||
</p>
|
||||
{scheduledPost.error ? (
|
||||
{scheduledPost.error && !scheduledPost.publishedAt && scheduledPost.nextAttemptAt ? (
|
||||
<Badge variant="warning" className="gap-1">
|
||||
<RotateCw className="h-3 w-3" /> Nouvel essai
|
||||
</Badge>
|
||||
) : scheduledPost.error ? (
|
||||
<Badge variant="danger" className="gap-1">
|
||||
<XCircle className="h-3 w-3" /> Échec
|
||||
</Badge>
|
||||
|
||||
@@ -2,7 +2,7 @@ import { useState } from "react";
|
||||
import { Page } from "@/components/layout/app-shell";
|
||||
import { PageHeader } from "@/components/layout/page-header";
|
||||
import { useQuery } from "@tanstack/react-query";
|
||||
import { CheckCircle, XCircle, Calendar, History as HistoryIcon, Eye, Image as ImageIcon, Smartphone, Loader2, Clapperboard, Search, Sparkles } from "lucide-react";
|
||||
import { CheckCircle, XCircle, RotateCw, Calendar, History as HistoryIcon, Eye, Image as ImageIcon, Smartphone, Loader2, Clapperboard, Search, Sparkles } from "lucide-react";
|
||||
import { Link } from "wouter";
|
||||
import { Card } from "@/components/ui/card";
|
||||
import { Input } from "@/components/ui/input";
|
||||
@@ -27,10 +27,17 @@ function isAwaitingPublication(scheduledPost: ScheduledPostWithRelations): boole
|
||||
return (
|
||||
!!scheduledPost.scheduledAt &&
|
||||
new Date(scheduledPost.scheduledAt) <= new Date() &&
|
||||
!scheduledPost.publishedAt
|
||||
!scheduledPost.publishedAt &&
|
||||
// Un échec ne bouge plus avant sa prochaine tentative (minutes ou heures)
|
||||
!scheduledPost.error
|
||||
);
|
||||
}
|
||||
|
||||
/** Échec dont le scheduler fera une nouvelle tentative automatique. */
|
||||
function isRetrying(scheduledPost: ScheduledPostWithRelations): boolean {
|
||||
return !!scheduledPost.error && !scheduledPost.publishedAt && !!scheduledPost.nextAttemptAt;
|
||||
}
|
||||
|
||||
type StatusFilter = "all" | "published" | "pending" | "failed";
|
||||
|
||||
const STATUS_TABS: { value: StatusFilter; label: string }[] = [
|
||||
@@ -115,8 +122,13 @@ export default function History() {
|
||||
return null;
|
||||
};
|
||||
|
||||
// Un échec avec un nouvel essai programmé est encore « en cours »
|
||||
const statusOf = (sp: ScheduledPostWithRelations): StatusFilter =>
|
||||
sp.error ? "failed" : !sp.publishedAt ? "pending" : "published";
|
||||
sp.publishedAt && !sp.error
|
||||
? "published"
|
||||
: sp.error && !isRetrying(sp)
|
||||
? "failed"
|
||||
: "pending";
|
||||
|
||||
const counts = publishedPosts.reduce(
|
||||
(acc, sp) => {
|
||||
@@ -200,6 +212,8 @@ export default function History() {
|
||||
<p className="truncate font-medium">{scheduledPost.page?.pageName || "Page inconnue"}</p>
|
||||
{status === "failed" ? (
|
||||
<Badge variant="danger" className="gap-1"><XCircle className="h-3 w-3" /> Échec</Badge>
|
||||
) : isRetrying(scheduledPost) ? (
|
||||
<Badge variant="warning" className="gap-1"><RotateCw className="h-3 w-3" /> Nouvel essai</Badge>
|
||||
) : status === "pending" ? (
|
||||
<Badge variant="info" className="gap-1"><Loader2 className="h-3 w-3 animate-spin" /> En cours</Badge>
|
||||
) : (
|
||||
@@ -233,6 +247,15 @@ export default function History() {
|
||||
{scheduledPost.error && (
|
||||
<p className="mt-2 rounded-md bg-destructive/10 px-2.5 py-1.5 text-xs text-destructive">{scheduledPost.error}</p>
|
||||
)}
|
||||
{scheduledPost.error && !scheduledPost.publishedAt && (
|
||||
<p className="mt-1 text-xs text-muted-foreground">
|
||||
{scheduledPost.nextAttemptAt
|
||||
? `Nouvelle tentative automatique le ${format(new Date(scheduledPost.nextAttemptAt), "d MMM yyyy 'à' HH:mm", { locale: fr })}`
|
||||
: scheduledPost.attempts > 0
|
||||
? "Plus de nouvelle tentative automatique : reprogrammez la publication pour réessayer."
|
||||
: null}
|
||||
</p>
|
||||
)}
|
||||
</div>
|
||||
<Button
|
||||
variant="ghost"
|
||||
|
||||
+12
-2
@@ -907,11 +907,21 @@ impl Db {
|
||||
}
|
||||
|
||||
/// Publications dues et pas encore traitées.
|
||||
pub async fn get_pending_scheduled_posts(&self) -> Result<Vec<ScheduledPost>, sqlx::Error> {
|
||||
/// Publications dues, hors échecs en attente de leur prochaine tentative
|
||||
/// et hors publications abandonnées après `max_attempts` échecs.
|
||||
pub async fn get_pending_scheduled_posts(
|
||||
&self,
|
||||
max_attempts: i32,
|
||||
) -> Result<Vec<ScheduledPost>, sqlx::Error> {
|
||||
sqlx::query_as(
|
||||
"SELECT * FROM scheduled_posts WHERE published_at IS NULL AND scheduled_at <= $1",
|
||||
r#"SELECT * FROM scheduled_posts
|
||||
WHERE published_at IS NULL AND scheduled_at <= $1
|
||||
AND (next_attempt_at IS NULL OR next_attempt_at <= $1)
|
||||
AND attempts < $2
|
||||
ORDER BY scheduled_at"#,
|
||||
)
|
||||
.bind(Ts::now())
|
||||
.bind(max_attempts)
|
||||
.fetch_all(&self.pool)
|
||||
.await
|
||||
}
|
||||
|
||||
@@ -175,6 +175,10 @@ pub struct ScheduledPost {
|
||||
pub error: Option<String>,
|
||||
pub publish_id: Option<String>,
|
||||
pub publish_status: Option<String>,
|
||||
/// Tentatives de publication en échec.
|
||||
pub attempts: i32,
|
||||
/// Prochaine tentative après un échec (`None` : immédiate ou abandonnée).
|
||||
pub next_attempt_at: Option<Ts>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, FromRow)]
|
||||
|
||||
@@ -251,6 +251,9 @@ const COLUMNS: &[&str] = &[
|
||||
r#"ALTER TABLE "scheduled_posts"
|
||||
ADD COLUMN IF NOT EXISTS "publish_id" text,
|
||||
ADD COLUMN IF NOT EXISTS "publish_status" text"#,
|
||||
r#"ALTER TABLE "scheduled_posts"
|
||||
ADD COLUMN IF NOT EXISTS "attempts" integer NOT NULL DEFAULT 0,
|
||||
ADD COLUMN IF NOT EXISTS "next_attempt_at" timestamp"#,
|
||||
];
|
||||
|
||||
/// Index de performance (Postgres n'indexe pas les clés étrangères seul).
|
||||
|
||||
@@ -214,6 +214,8 @@ mod tests {
|
||||
error: error.map(str::to_string),
|
||||
publish_id: None,
|
||||
publish_status: None,
|
||||
attempts: 0,
|
||||
next_attempt_at: None,
|
||||
},
|
||||
post: ScheduledPostPost {
|
||||
id: "p".into(),
|
||||
|
||||
@@ -160,7 +160,12 @@ async fn update_scheduled(
|
||||
let mut patch = Patch::new();
|
||||
if let Some(value) = body.get("scheduledAt").filter(|v| is_truthy(Some(v))) {
|
||||
let date = js_date(value).ok_or_else(|| AppError::internal(FAILED))?;
|
||||
patch = patch.set("scheduled_at", date);
|
||||
// Reprogrammer relance les publications en échec ou abandonnées
|
||||
patch = patch
|
||||
.set("scheduled_at", date)
|
||||
.set("attempts", 0)
|
||||
.set("next_attempt_at", None::<Ts>)
|
||||
.set_null("error");
|
||||
state
|
||||
.db
|
||||
.update_post(&scheduled.post_id, Patch::new().set("scheduled_for", date))
|
||||
|
||||
@@ -869,7 +869,14 @@ async fn update_post_inner(state: &AppState, id: &str, body: &Value) -> Outcome
|
||||
if sp.published_at.is_none() {
|
||||
state
|
||||
.db
|
||||
.update_scheduled_post(&sp.id, Patch::new().set("scheduled_at", new_date))
|
||||
.update_scheduled_post(
|
||||
&sp.id,
|
||||
Patch::new()
|
||||
.set("scheduled_at", new_date)
|
||||
.set("attempts", 0)
|
||||
.set("next_attempt_at", None::<Ts>)
|
||||
.set_null("error"),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,9 +1,13 @@
|
||||
//! Publication des posts programmés : toutes les minutes, les publications
|
||||
//! dues sont envoyées sur Facebook ou TikTok.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::AtomicBool;
|
||||
use std::sync::Arc;
|
||||
|
||||
use once_cell::sync::Lazy;
|
||||
use regex::Regex;
|
||||
|
||||
use crate::db::{
|
||||
Media, MediaType, Patch, Platform, Post, PostStatus, PostType, ScheduledPost, SocialPage,
|
||||
};
|
||||
@@ -29,9 +33,78 @@ pub fn start(state: AppState) {
|
||||
tracing::info!("Scheduler service started");
|
||||
}
|
||||
|
||||
/// Nombre de tentatives en échec après lequel une publication est abandonnée
|
||||
/// (il faut alors la reprogrammer).
|
||||
pub const MAX_PUBLISH_ATTEMPTS: i32 = 5;
|
||||
|
||||
/// Retard au-delà duquel une publication n'est plus envoyée : une promotion
|
||||
/// ou une story publiée un jour trop tard fait plus de mal que de bien.
|
||||
const MAX_PUBLISH_DELAY_HOURS: i64 = 24;
|
||||
|
||||
/// Publication trop en retard sur son heure prévue pour être encore envoyée.
|
||||
fn is_stale(scheduled_at: Ts, now: Ts) -> bool {
|
||||
scheduled_at.add(chrono::Duration::hours(MAX_PUBLISH_DELAY_HOURS)) < now
|
||||
}
|
||||
|
||||
/// Délai avant la tentative suivante, selon le nombre d'échecs déjà subis.
|
||||
/// Une limite de débit (quota d'appels, blocage anti-spam) appelle des délais
|
||||
/// plus longs : réessayer tout de suite prolonge le blocage.
|
||||
fn retry_delay(attempts: i32, rate_limited: bool) -> chrono::Duration {
|
||||
const MINUTES: [i64; 4] = [5, 15, 60, 180];
|
||||
const RATE_LIMITED_MINUTES: [i64; 4] = [30, 60, 120, 240];
|
||||
let table = if rate_limited {
|
||||
&RATE_LIMITED_MINUTES
|
||||
} else {
|
||||
&MINUTES
|
||||
};
|
||||
let index = usize::try_from(attempts - 1)
|
||||
.unwrap_or(0)
|
||||
.min(table.len() - 1);
|
||||
chrono::Duration::minutes(table[index])
|
||||
}
|
||||
|
||||
/// Code d'erreur Graph repris dans le message (`... (code: 4)`).
|
||||
fn graph_error_code(message: &str) -> Option<i64> {
|
||||
static CODE: Lazy<Regex> = Lazy::new(|| Regex::new(r"\(code: (\d+)").unwrap());
|
||||
CODE.captures(message)?.get(1)?.as_str().parse().ok()
|
||||
}
|
||||
|
||||
/// Erreur due à une limite de débit : quotas de l'application (4), de
|
||||
/// l'utilisateur (17), de la page (32, 613, 80001…) ou blocage temporaire
|
||||
/// anti-spam (368), ou `rate_limit_exceeded` côté TikTok.
|
||||
fn is_rate_limited(message: &str) -> bool {
|
||||
if let Some(code) = graph_error_code(message) {
|
||||
if matches!(code, 4 | 17 | 32 | 368 | 613 | 80001..=80014) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
let lower = message.to_lowercase();
|
||||
lower.contains("rate_limit") || lower.contains("request limit")
|
||||
}
|
||||
|
||||
async fn page_platform(state: &AppState, page_id: &str) -> Option<Platform> {
|
||||
match state.db.get_social_page(page_id).await {
|
||||
Ok(page) => page.map(|p| p.platform),
|
||||
Err(error) => {
|
||||
tracing::warn!("[Scheduler] Page {page_id} illisible : {error}");
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Traite les publications dues (une erreur n'interrompt pas les suivantes).
|
||||
///
|
||||
/// Un échec n'est retenté qu'après un délai croissant, et au plus
|
||||
/// `MAX_PUBLISH_ATTEMPTS` fois : sans cela, chaque publication en échec était
|
||||
/// renvoyée toutes les minutes et épuisait le quota d'appels de l'application.
|
||||
/// Après une limite de débit, les autres publications de la même plateforme
|
||||
/// sont reportées au lieu d'aggraver le blocage.
|
||||
pub async fn process_pending_posts(state: &AppState) {
|
||||
let pending = match state.db.get_pending_scheduled_posts().await {
|
||||
let pending = match state
|
||||
.db
|
||||
.get_pending_scheduled_posts(MAX_PUBLISH_ATTEMPTS)
|
||||
.await
|
||||
{
|
||||
Ok(pending) => pending,
|
||||
Err(error) => {
|
||||
tracing::error!("Error processing pending posts: {error}");
|
||||
@@ -39,14 +112,93 @@ pub async fn process_pending_posts(state: &AppState) {
|
||||
}
|
||||
};
|
||||
|
||||
let mut cooldowns: HashMap<Platform, Ts> = HashMap::new();
|
||||
|
||||
for scheduled_post in pending {
|
||||
if is_stale(scheduled_post.scheduled_at, Ts::now()) {
|
||||
tracing::warn!(
|
||||
"[Scheduler] Post {} abandonné : plus de {MAX_PUBLISH_DELAY_HOURS} h de retard",
|
||||
scheduled_post.id
|
||||
);
|
||||
// Le dernier message d'erreur est conservé s'il y en a un
|
||||
let reason = format!(
|
||||
"Publication abandonnée : plus de {MAX_PUBLISH_DELAY_HOURS} h de retard sur l'heure prévue."
|
||||
);
|
||||
let error = match scheduled_post.error.as_deref().filter(|e| !e.is_empty()) {
|
||||
Some(last) => format!("{reason} Dernière erreur : {last}"),
|
||||
None => reason,
|
||||
};
|
||||
let abandoned = state
|
||||
.db
|
||||
.update_scheduled_post(
|
||||
&scheduled_post.id,
|
||||
Patch::new()
|
||||
.set("error", error)
|
||||
.set("attempts", MAX_PUBLISH_ATTEMPTS)
|
||||
.set("next_attempt_at", None::<Ts>),
|
||||
)
|
||||
.await;
|
||||
if let Err(error) = abandoned {
|
||||
tracing::error!("Error processing pending posts: {error}");
|
||||
return;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if !cooldowns.is_empty() {
|
||||
let platform = page_platform(state, &scheduled_post.page_id).await;
|
||||
if let Some(until) = platform.and_then(|p| cooldowns.get(&p)).copied() {
|
||||
tracing::info!(
|
||||
"[Scheduler] Post {} reporté au {until} (limite de débit atteinte)",
|
||||
scheduled_post.id
|
||||
);
|
||||
let deferred = state
|
||||
.db
|
||||
.update_scheduled_post(
|
||||
&scheduled_post.id,
|
||||
Patch::new().set("next_attempt_at", until),
|
||||
)
|
||||
.await;
|
||||
if let Err(error) = deferred {
|
||||
tracing::error!("Error processing pending posts: {error}");
|
||||
return;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
if let Err(error) = process_one(state, &scheduled_post).await {
|
||||
tracing::error!("Error publishing post {}: {error:#}", scheduled_post.id);
|
||||
let message = error_message(&error);
|
||||
let rate_limited = is_rate_limited(&message);
|
||||
let attempts = scheduled_post.attempts + 1;
|
||||
let retry_at = Ts::now().add(retry_delay(attempts, rate_limited));
|
||||
let next_attempt_at = (attempts < MAX_PUBLISH_ATTEMPTS).then_some(retry_at);
|
||||
|
||||
match next_attempt_at {
|
||||
Some(at) => tracing::error!(
|
||||
"Error publishing post {} (tentative {attempts}/{MAX_PUBLISH_ATTEMPTS}, nouvel essai le {at}): {error:#}",
|
||||
scheduled_post.id
|
||||
),
|
||||
None => tracing::error!(
|
||||
"Error publishing post {} (abandon après {attempts} tentatives): {error:#}",
|
||||
scheduled_post.id
|
||||
),
|
||||
}
|
||||
|
||||
if rate_limited {
|
||||
if let Some(platform) = page_platform(state, &scheduled_post.page_id).await {
|
||||
cooldowns.insert(platform, retry_at);
|
||||
}
|
||||
}
|
||||
|
||||
let recorded = state
|
||||
.db
|
||||
.update_scheduled_post(
|
||||
&scheduled_post.id,
|
||||
Patch::new().set("error", error_message(&error)),
|
||||
Patch::new()
|
||||
.set("error", message)
|
||||
.set("attempts", attempts)
|
||||
.set("next_attempt_at", next_attempt_at),
|
||||
)
|
||||
.await;
|
||||
if let Err(error) = recorded {
|
||||
@@ -136,7 +288,8 @@ async fn publish_post(state: &AppState, scheduled_post: &ScheduledPost) -> anyho
|
||||
Patch::new()
|
||||
.set("published_at", Ts::now())
|
||||
.set("external_post_id", external_post_id)
|
||||
.set_null("error"),
|
||||
.set_null("error")
|
||||
.set("next_attempt_at", None::<Ts>),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
@@ -154,7 +307,8 @@ async fn publish_post(state: &AppState, scheduled_post: &ScheduledPost) -> anyho
|
||||
.set("published_at", Ts::now())
|
||||
.set("publish_id", publish_id)
|
||||
.set("publish_status", "PROCESSING_UPLOAD")
|
||||
.set_null("error"),
|
||||
.set_null("error")
|
||||
.set("next_attempt_at", None::<Ts>),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
@@ -254,3 +408,51 @@ async fn publish_to_tiktok(
|
||||
|
||||
tiktok::publish_video_from_buffer(state, page, buffer, &post.content).await
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn detects_facebook_rate_limits() {
|
||||
assert!(is_rate_limited(
|
||||
"Facebook Story API error (START phase): (#4) Application request limit reached (code: 4)"
|
||||
));
|
||||
assert!(is_rate_limited(
|
||||
"Facebook Reel START error: Nous limitons le nombre de fois que vous pouvez publier (code: 368, subcode: 1390008)"
|
||||
));
|
||||
assert!(is_rate_limited(
|
||||
"Facebook API error: Page request limit reached (code: 32)"
|
||||
));
|
||||
assert!(is_rate_limited("TikTok error: rate_limit_exceeded"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn other_errors_are_not_rate_limits() {
|
||||
assert!(!is_rate_limited(
|
||||
"Facebook API error: Error validating access token (code: 190)"
|
||||
));
|
||||
assert!(!is_rate_limited(
|
||||
"Facebook API error: Invalid parameter (code: 100)"
|
||||
));
|
||||
assert!(!is_rate_limited("Post or page not found"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn posts_more_than_a_day_late_are_stale() {
|
||||
let now = Ts::parse("2026-10-12T19:00:00Z").unwrap();
|
||||
assert!(is_stale(Ts::parse("2026-10-11T16:30:00Z").unwrap(), now));
|
||||
assert!(!is_stale(Ts::parse("2026-10-12T16:30:00Z").unwrap(), now));
|
||||
assert!(!is_stale(Ts::parse("2026-10-11T19:00:00Z").unwrap(), now));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn retry_delays_grow_and_stay_bounded() {
|
||||
assert_eq!(retry_delay(1, false), chrono::Duration::minutes(5));
|
||||
assert_eq!(retry_delay(2, false), chrono::Duration::minutes(15));
|
||||
assert_eq!(retry_delay(9, false), chrono::Duration::minutes(180));
|
||||
assert_eq!(retry_delay(1, true), chrono::Duration::minutes(30));
|
||||
assert_eq!(retry_delay(4, true), chrono::Duration::minutes(240));
|
||||
assert_eq!(retry_delay(0, true), chrono::Duration::minutes(30));
|
||||
}
|
||||
}
|
||||
@@ -230,6 +230,11 @@ export const scheduledPosts = pgTable("scheduled_posts", {
|
||||
// post_id définitif n'est connu qu'après interrogation du statut.
|
||||
publishId: text("publish_id"),
|
||||
publishStatus: text("publish_status"),
|
||||
|
||||
// Échecs de publication : nouvel essai à `nextAttemptAt` (délai croissant),
|
||||
// abandon après quelques tentatives (il faut alors reprogrammer).
|
||||
attempts: integer("attempts").notNull().default(0),
|
||||
nextAttemptAt: timestamp("next_attempt_at"),
|
||||
});
|
||||
|
||||
export const aiGenerations = pgTable("ai_generations", {
|
||||
@@ -454,6 +459,8 @@ export const insertScheduledPostSchema = createInsertSchema(scheduledPosts).omit
|
||||
publishedAt: true,
|
||||
externalPostId: true,
|
||||
error: true,
|
||||
attempts: true,
|
||||
nextAttemptAt: true,
|
||||
});
|
||||
|
||||
export const insertAiGenerationSchema = createInsertSchema(aiGenerations).omit({
|
||||
|
||||
Reference in new issue
Block a user