"""Launchpad: publishes due calendar items (with the images or video of the artifact they were made from), emails reminders for platforms without an API, rewrites links to tracked short links and pulls engagement back in. Due items are published by a "publish_post" job each, not inside the worker's 30 second tick: a video upload and the platform's processing can take minutes.""" import logging import re import secrets import uuid from collections.abc import Callable from datetime import UTC, datetime, timedelta from sqlalchemy import func, select from sqlalchemy.orm import Session from app.config import settings from app.db import SessionLocal from app.models import Artifact, CalendarItem, Document, EngagementEvent, SocialAccount, TrackedLink, User, Workspace from app.services.email import email_service from app.services.jobs import create_job from app.services.social import AUTO_POST, PLATFORM_LABELS, SocialError, module from app.services.social.health import mark_down from app.services.social.media import carries_media, media_count, media_for log = logging.getLogger("notestack.launchpad") _URL = re.compile(r"https?://[^\s)\]>]+") MAX_ATTEMPTS = 3 STUCK_AFTER = timedelta(minutes=30) # longer than a video upload and the platform's processing take def _public_api() -> bool: return not re.search(r"//(localhost|127\.0\.0\.1|api)(:|/|$)", settings.api_url) def track_links(db: Session, item: CalendarItem, text: str) -> str: """Swap outbound links for {API_URL}/l/{slug} so clicks are counted. Skipped when the API is not public.""" if not _public_api(): return text base = f"{settings.api_url.rstrip('/')}/l/" def swap(m: re.Match) -> str: url = m.group(0) if url.startswith(base): return url link = TrackedLink(workspace_id=item.workspace_id, slug=secrets.token_urlsafe(5)[:7], target_url=url, calendar_item_id=item.id) db.add(link) return base + link.slug return _URL.sub(swap, text) def owner_email(db: Session, workspace_id) -> str | None: ws = db.get(Workspace, workspace_id) user = db.get(User, ws.owner_id) if ws else None return user.email if user and user.is_active else None def publish_item(db: Session, item: CalendarItem) -> CalendarItem: """Publish or remind now. Leaves the item posted, reminded, failed, or scheduled again for a retry.""" posts = [p for p in [item.content, *(item.thread or [])] if p and p.strip()] if not posts and carries_media(db.get(Artifact, item.artifact_id) if item.artifact_id else None): posts = [""] # a video or images on their own: the post goes out without a caption if not posts: item.status, item.error = "failed", "Nothing to post." db.commit() return item account = db.get(SocialAccount, item.social_account_id) if item.social_account_id else None label = PLATFORM_LABELS.get(item.platform, item.platform) if item.platform in ("x", "linkedin") and not item.remind_by_email and (not account or account.status != "active"): # The connection is down: never attempted; it resumes when the account is reconnected. item.status, item.error = "paused", f"{label} disconnected. Reconnect {label} to resume." db.commit() return item if item.platform not in AUTO_POST or item.remind_by_email or not account: to = owner_email(db, item.workspace_id) sent = bool(to) and email_service.send_post_reminder( to, label, "\n\n".join(posts), f"{settings.frontend_url}/app/launchpad?item={item.id}") item.status = "reminded" if sent else "failed" item.error = None if sent else "Could not send the reminder email." item.posted_at = datetime.now(UTC) db.commit() return item item.attempts += 1 try: tracked = [track_links(db, item, p) for p in posts] db.flush() artifact = db.get(Artifact, item.artifact_id) if item.artifact_id else None if artifact and media_count(artifact) and item.platform in ("x", "linkedin") \ and not module(item.platform).can_post_media(account): # Scheduled before the account lost the right to post media: held, not attempted, until reconnected. item.status, item.error = "paused", f"Reconnect {label} to post images and videos." db.commit() return item with media_for(artifact, item.platform) as media: # temporary video files go when this ends external_id, url = module(item.platform).publish(db, account, tracked, media) except SocialError as exc: if exc.reconnect: # the platform refused the connection: it is down, and its posts pause mark_down(db, account, "expired", str(exc)) item.status, item.error = "paused", str(exc)[:1000] db.commit() return item _fail_or_retry(db, item, str(exc), exc.permanent) return item except Exception as exc: # network trouble: retry log.exception("publish failed for %s", item.id) _fail_or_retry(db, item, f"{type(exc).__name__}: {exc}", False) return item item.status, item.error = "posted", None item.external_id, item.external_url, item.posted_at = external_id, url, datetime.now(UTC) if item.document_id: doc = db.get(Document, item.document_id) if doc: doc.last_resurfaced_at = item.posted_at db.commit() return item def _fail_or_retry(db: Session, item: CalendarItem, error: str, permanent: bool) -> None: item.error = error[:1000] if permanent or item.attempts >= MAX_ATTEMPTS: item.status = "failed" else: item.status = "scheduled" item.scheduled_at = datetime.now(UTC) + timedelta(minutes=5 * item.attempts) db.commit() def has_media(db: Session, item: CalendarItem) -> bool: """The post uploads images or a video (slow: it goes through a job, also for Publish now).""" artifact = db.get(Artifact, item.artifact_id) if item.artifact_id else None return bool(artifact) and item.platform in ("x", "linkedin") and media_count(artifact) > 0 def queue_publish(db: Session, item: CalendarItem) -> None: """Hand the (already claimed, "publishing") item to a worker job.""" create_job(db, item.workspace_id, "publish_post", {"item_id": str(item.id)}, max_attempts=1) def publish_job(db: Session, item_id: str) -> CalendarItem | None: """The publish_post job: publish the item if it is still waiting for it.""" item = db.get(CalendarItem, uuid.UUID(str(item_id))) if not item or item.status != "publishing": return item item.updated_at = datetime.now(UTC) # the stuck limit counts from the job's start, not the claim db.commit() return publish_item(db, item) def publish_due(session_factory: Callable[[], Session] = SessionLocal) -> int: """Claim the due items and queue a publish_post job for each. Returns how many were queued.""" now = datetime.now(UTC) with session_factory() as db: # A job that died mid publish leaves its item in "publishing"; put it back once that has clearly happened. for stuck in db.scalars(select(CalendarItem).where( CalendarItem.status == "publishing", CalendarItem.updated_at < now - STUCK_AFTER)): stuck.status = "scheduled" due = db.scalars( select(CalendarItem).where(CalendarItem.status == "scheduled", CalendarItem.scheduled_at <= now) .order_by(CalendarItem.scheduled_at).limit(50) ).all() for item in due: item.status = "publishing" # claim, so an overlapping tick skips it db.commit() for item in due: queue_publish(db, item) return len(due) def clicks_by_item(db: Session, item_ids: list) -> dict: if not item_ids: return {} rows = db.execute( select(TrackedLink.calendar_item_id, func.count(EngagementEvent.id)) .join(EngagementEvent, EngagementEvent.tracked_link_id == TrackedLink.id) .where(TrackedLink.calendar_item_id.in_(item_ids), EngagementEvent.kind == "click") .group_by(TrackedLink.calendar_item_id) ).all() return {item_id: n for item_id, n in rows} def sync_engagement(session_factory: Callable[[], Session] = SessionLocal) -> int: since = datetime.now(UTC) - timedelta(days=30) updated = 0 with session_factory() as db: items = db.scalars( select(CalendarItem).where(CalendarItem.status == "posted", CalendarItem.posted_at >= since, CalendarItem.external_id.is_not(None)) ).all() for item in items: account = db.get(SocialAccount, item.social_account_id) if item.social_account_id else None if not account: continue try: data = module(item.platform).metrics(db, account, item.external_id) except Exception: log.warning("metrics failed for %s", item.id, exc_info=True) continue if data: item.metrics = {**(item.metrics or {}), **data, "synced_at": datetime.now(UTC).isoformat()} updated += 1 db.commit() return updated