diff --git a/alembic/versions/ee6174eeef14_service_notifications.py b/alembic/versions/ee6174eeef14_service_notifications.py new file mode 100644 index 0000000..17d551c --- /dev/null +++ b/alembic/versions/ee6174eeef14_service_notifications.py @@ -0,0 +1,47 @@ +"""+service_notifications + +Revision ID: ee6174eeef14 +Revises: ad537d63c440 +Create Date: 2026-08-20 12:01:51.073160 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'ee6174eeef14' +down_revision: Union[str, Sequence[str], None] = 'ad537d63c440' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('service_notifications', + sa.Column('id', sa.INTEGER(), autoincrement=True, nullable=False), + sa.Column('subscription_id', sa.INTEGER(), nullable=False), + sa.Column('notify_type', sa.Enum('SEVEN_DAYS', 'THREE_DAYS', 'ONE_DAY', 'EXPIRED', name='notificationtype'), nullable=False), + sa.Column('sub_expires_at', sa.DateTime(timezone=True), nullable=False), + sa.Column('status', sa.Enum('PENDING', 'DISPATCHED', 'SENT', 'FAILED', name='notificationstatus'), nullable=False), + sa.Column('dispatched_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('sent_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('attempts', sa.INTEGER(), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(['subscription_id'], ['subscriptions.id'], ), + sa.PrimaryKeyConstraint('id'), + sa.UniqueConstraint('id'), + sa.UniqueConstraint('sub_expires_at'), + sa.UniqueConstraint('subscription_id', 'notify_type', 'sub_expires_at', name='uq_subscription_notification') + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_table('service_notifications') + # ### end Alembic commands ### diff --git a/core/secrets.py b/core/secrets.py index f85fb08..c6e0e29 100644 --- a/core/secrets.py +++ b/core/secrets.py @@ -37,7 +37,11 @@ def generate_jwt(payload: dict[str, Any]) -> str: def get_kid_from_token(token: str) -> str | None: - return jwt.get_unverified_header(token).get("kid") + try: + header = jwt.get_unverified_header(token) + return header.get("kid") + except jwt.exceptions.PyJWTError: + return def decode_jwt( diff --git a/db/models/__init__.py b/db/models/__init__.py index 98b1087..12822f7 100644 --- a/db/models/__init__.py +++ b/db/models/__init__.py @@ -3,6 +3,7 @@ from .invoice import Invoice from .link_codes import LinkCode from .orders import Order, OrderAddon from .pricing import PricingConfig +from .service_notifications import ServiceNotification from .service_signatures import ServiceSignature from .sessions import Session from .subscription_addons import SubscriptionAddon @@ -18,6 +19,7 @@ __all__ = [ "Order", "OrderAddon", "PricingConfig", + "ServiceNotification", "ServiceSignature", "Session", "Subscription", diff --git a/db/models/service_notifications.py b/db/models/service_notifications.py new file mode 100644 index 0000000..8dfd5d0 --- /dev/null +++ b/db/models/service_notifications.py @@ -0,0 +1,54 @@ +from datetime import UTC, datetime +from typing import TYPE_CHECKING + +from sqlalchemy import INTEGER, DateTime, Enum, ForeignKey, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from db.base import Base +from schemas.enums import NotificationStatus, NotificationType + +if TYPE_CHECKING: + from db.models import Subscription + + +class ServiceNotification(Base): + __tablename__ = "service_notifications" + + id: Mapped[int] = mapped_column( + INTEGER, unique=True, autoincrement=True, primary_key=True, nullable=False + ) + subscription_id: Mapped[int] = mapped_column( + ForeignKey("subscriptions.id"), nullable=False, unique=False + ) + notify_type: Mapped[NotificationType] = mapped_column( + Enum(NotificationType, name="notificationtype"), nullable=False + ) + sub_expires_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), unique=True, nullable=False + ) + + status: Mapped[NotificationStatus] = mapped_column( + Enum(NotificationStatus, name="notificationstatus"), + nullable=False, + default=NotificationStatus.PENDING, + ) + dispatched_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=True) + sent_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=True) + attempts: Mapped[int] = mapped_column(INTEGER, nullable=False, default=0) + + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + nullable=False, + default=lambda: datetime.now(UTC), + ) + + __table_args__ = ( + UniqueConstraint( + "subscription_id", + "notify_type", + "sub_expires_at", + name="uq_subscription_notification", + ), + ) + + subscription: Mapped["Subscription"] = relationship("Subscription", lazy="selectin") diff --git a/repositories/service_notifications.py b/repositories/service_notifications.py new file mode 100644 index 0000000..4c497c0 --- /dev/null +++ b/repositories/service_notifications.py @@ -0,0 +1,64 @@ +from sqlalchemy import func, or_, select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from db.models.service_notifications import ServiceNotification +from schemas.enums import NotificationStatus + + +async def get_notification_by_id(session: AsyncSession, n_id: int) -> ServiceNotification | None: + stmt = select(ServiceNotification).where(ServiceNotification.id == n_id) + r = await session.execute(stmt) + + return r.scalar_one_or_none() + + +async def get_pending_notifications( + session: AsyncSession, batch_size: int = 50 +) -> list[ServiceNotification]: + stmt = ( + select(ServiceNotification) + .where( + or_( + ServiceNotification.status == "pending", + ( + (ServiceNotification.status == "dispatched") + & ( + ServiceNotification.dispatched_at + < func.now() - text("interval '10 minutes'") + ) + ), + ) + ) + .order_by(ServiceNotification.created_at) + .with_for_update(skip_locked=True) + .limit(batch_size) + ) + r = await session.execute(stmt) + + return list(r.scalars().all()) + + +async def ack_notification(session: AsyncSession, n_id: int) -> ServiceNotification | None: + notification = await get_notification_by_id(session, n_id) + if not notification: + return + + notification.sent_at = func.now() + notification.status = NotificationStatus.SENT + + await session.commit() + + return notification + + +async def mark_notification_as_dispatched(session: AsyncSession, n_id: int): + notification = await get_notification_by_id(session, n_id) + if not notification: + return + + notification.dispatched_at = func.now() + notification.status = NotificationStatus.DISPATCHED + notification.attempts += 1 + + await session.commit() + return notification diff --git a/routes/__init__.py b/routes/__init__.py index d2815ee..cec608f 100644 --- a/routes/__init__.py +++ b/routes/__init__.py @@ -2,6 +2,7 @@ from fastapi import APIRouter from .auth import router as auth_router from .health import router as health_router +from .internal import internal_router from .link_codes import router as link_code_router from .orders import router as orders_router from .payments import payment_router @@ -16,4 +17,5 @@ routers: list[APIRouter] = [ health_router, link_code_router, payment_router, + internal_router, ] diff --git a/routes/internal/__init__.py b/routes/internal/__init__.py new file mode 100644 index 0000000..dca0f25 --- /dev/null +++ b/routes/internal/__init__.py @@ -0,0 +1,6 @@ +from fastapi import APIRouter + +from .renewal import router as renewal_router + +internal_router = APIRouter(prefix="/internal") +internal_router.include_router(renewal_router) diff --git a/routes/internal/renewal.py b/routes/internal/renewal.py new file mode 100644 index 0000000..b50235f --- /dev/null +++ b/routes/internal/renewal.py @@ -0,0 +1,37 @@ +from fastapi import APIRouter, Depends +from sqlalchemy.ext.asyncio import AsyncSession + +from core.deps import get_service_identity +from db.session import get_db +from repositories.service_notifications import ( + get_pending_notifications, + mark_notification_as_dispatched, +) +from schemas.dto import ServiceIdentity +from schemas.notifications import NotificationResponse, UserNotificationData + +router = APIRouter(prefix="/renewal") + + +@router.get("/pending", response_model=NotificationResponse) +async def get_pending( + limit: int = 50, + ctx: ServiceIdentity = Depends(get_service_identity), + session: AsyncSession = Depends(get_db), +): + notifications = await get_pending_notifications(session, limit) + + response_users: list[UserNotificationData] = [] + for notification in notifications: + response_users.append( + UserNotificationData( + username=notification.subscription.user.username, + telegram_id=notification.subscription.user.telegram_id, + expires_at=notification.sub_expires_at, + notification_type=notification.notify_type, + ) + ) + + await mark_notification_as_dispatched(session, notification.id) + + return NotificationResponse(users=response_users, issued_by=ctx.service) diff --git a/schemas/enums.py b/schemas/enums.py index bc48fb0..8594c54 100644 --- a/schemas/enums.py +++ b/schemas/enums.py @@ -15,3 +15,17 @@ class LinkCodeStatus(StrEnum): ACTIVE = "active" USED = "used" EXPIRED = "expired" + + +class NotificationType(StrEnum): + SEVEN_DAYS = "7d" + THREE_DAYS = "3d" + ONE_DAY = "1d" + EXPIRED = "expired" + + +class NotificationStatus(StrEnum): + PENDING = "pending" + DISPATCHED = "dispatched" + SENT = "sent" + FAILED = "failed" diff --git a/schemas/notifications.py b/schemas/notifications.py new file mode 100644 index 0000000..ec4fa22 --- /dev/null +++ b/schemas/notifications.py @@ -0,0 +1,19 @@ +from datetime import UTC, datetime + +from pydantic import BaseModel, Field + +from schemas.enums import NotificationType + + +class UserNotificationData(BaseModel): + username: str = Field() + telegram_id: int | None = Field() + expires_at: datetime = Field() + notification_type: NotificationType = Field() + + +class NotificationResponse(BaseModel): + created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + issued_by: str = Field() + + users: list[UserNotificationData] = Field()