[] = [
+ {
+ accessorKey: "subject",
+ header: ({ column }) => {
+ return (
+
+ );
+ },
+ cell: ({ row }) => (
+
+ {row.getValue("subject")}
+
+ ),
+ },
+ {
+ accessorKey: "status",
+ header: "Status",
+ cell: ({ row }) => {
+ const status = row.getValue("status") as BroadcastStatus;
+ return (
+
+ {status.charAt(0).toUpperCase() + status.slice(1)}
+
+ );
+ },
+ },
+ {
+ accessorKey: "recipient_type",
+ header: "Recipients",
+ cell: ({ row }) => {
+ const broadcast = row.original;
+ const type = broadcast.recipient_type;
+ if (type === "channel_members") {
+ return (
+
+ {broadcast.channel?.emoji} {broadcast.channel?.name || "Channel"}
+
+ );
+ }
+ return (
+
+ {type === "all_users" ? "All Users" : "Test Email"}
+
+ );
+ },
+ },
+ {
+ accessorKey: "sent_count",
+ header: "Sent",
+ cell: ({ row }) => {
+ const broadcast = row.original;
+ return (
+
+ {broadcast.sent_count > 0 || broadcast.failed_count > 0 ? (
+
+ {broadcast.sent_count} /{" "}
+ {broadcast.sent_count + broadcast.failed_count}
+
+ ) : (
+ "-"
+ )}
+
+ );
+ },
+ },
+ {
+ accessorKey: "created_at",
+ header: ({ column }) => {
+ return (
+
+ );
+ },
+ cell: ({ row }) => {
+ const dateValue = row.getValue("created_at") as string;
+ if (!dateValue) return N/A
;
+ const date = new Date(dateValue);
+ return (
+
+ {Number.isNaN(date.getTime())
+ ? "Invalid date"
+ : format(date, "MMM dd, yyyy")}
+
+ );
+ },
+ },
+ {
+ id: "actions",
+ header: "Actions",
+ cell: ({ row }) => {
+ const broadcast = row.original;
+ return (
+
+
+
+
+ {broadcast.status === "draft" && (
+
+
+
+ )}
+
+ );
+ },
+ },
+ ];
+
+ const filteredBroadcasts = useMemo(() => {
+ if (!searchQuery.trim()) return broadcasts;
+
+ const query = searchQuery.toLowerCase();
+ return broadcasts.filter((broadcast) => {
+ return broadcast.subject?.toLowerCase().includes(query);
+ });
+ }, [broadcasts, searchQuery]);
+
+ const table = useReactTable({
+ data: filteredBroadcasts,
+ columns,
+ getCoreRowModel: getCoreRowModel(),
+ getSortedRowModel: getSortedRowModel(),
+ getFilteredRowModel: getFilteredRowModel(),
+ onSortingChange: setSorting,
+ state: {
+ sorting,
+ },
+ });
+
+ if (loading) {
+ return ;
+ }
+
+ return (
+
+
+
+
+
+
setSearchQuery(e.target.value)}
+ className="pl-10"
+ />
+
+
+ {searchQuery && (
+
+ Showing {filteredBroadcasts.length} of {broadcasts.length} broadcasts
+
+ )}
+
+
+
+
+
+ {table.getHeaderGroups().map((headerGroup) => (
+
+ {headerGroup.headers.map((header) => (
+ |
+ {header.isPlaceholder
+ ? null
+ : flexRender(
+ header.column.columnDef.header,
+ header.getContext(),
+ )}
+ |
+ ))}
+
+ ))}
+
+
+ {table.getRowModel().rows?.length ? (
+ table.getRowModel().rows.map((row) => (
+
+ {row.getVisibleCells().map((cell) => (
+ |
+ {flexRender(
+ cell.column.columnDef.cell,
+ cell.getContext(),
+ )}
+ |
+ ))}
+
+ ))
+ ) : (
+
+ |
+ No broadcasts found.
+ |
+
+ )}
+
+
+
+
+
+
+
+
{broadcast.subject}
+
+
+ {broadcast.status.charAt(0).toUpperCase() +
+ broadcast.status.slice(1)}
+
+
+ Created {format(new Date(broadcast.created_at), "MMM dd, yyyy")}
+
+ {broadcast.sent_at && (
+
+ Sent {format(new Date(broadcast.sent_at), "MMM dd, yyyy HH:mm")}
+
+ )}
+
+
+
+
+
+
+
+
+ {broadcast.status === "draft" && (
+ <>
+
+
+
+ Send Test Email
+
+
+ setTestEmail(e.target.value)}
+ />
+
+
+
+
+
+
+ {broadcast.recipient_type === "channel_members" ? (
+
+ ) : (
+
+ )}
+ {broadcast.recipient_type === "channel_members"
+ ? `Send to ${broadcast.channel?.name || "Channel"} Members`
+ : "Send to All Users"}
+
+
+
+ {broadcast.recipient_type === "channel_members"
+ ? `This will send the broadcast to all members of ${broadcast.channel?.emoji || ""} ${broadcast.channel?.name || "the selected channel"}.`
+ : "This will send the broadcast to all active users."}
+
+
+
+ )}
+
+ {(broadcast.sent_count > 0 || broadcast.failed_count > 0) && (
+
+
Delivery Stats
+
+
+ Sent
+
+ {broadcast.sent_count}
+
+
+
+ Failed
+
+ {broadcast.failed_count}
+
+
+
+
+ )}
+
+
+
+
+
+
+
+ {[1, 2, 3, 4, 5].map((col) => (
+ |
+
+ |
+ ))}
+
+
+
+ {Array.from({ length: rowCount }).map((_, rowIndex) => (
+
+ {[1, 2, 3, 4, 5].map((col) => (
+ |
+
+ |
+ ))}
+
+ ))}
+
+
+
+
Broadcast
;
-}
diff --git a/apps/admin/src/routes/_dashboardLayout/broadcast/$id.tsx b/apps/admin/src/routes/_dashboardLayout/broadcast/$id.tsx
new file mode 100644
index 0000000..9c7d228
--- /dev/null
+++ b/apps/admin/src/routes/_dashboardLayout/broadcast/$id.tsx
@@ -0,0 +1,113 @@
+import { Button } from "@opencircle/ui";
+import { createFileRoute, Link, useNavigate } from "@tanstack/react-router";
+import { ArrowLeft } from "lucide-react";
+import { METADATA } from "../../../constants/metadata";
+import { BroadcastView } from "../../../features/broadcast/components/broadcastView";
+import { useBroadcastSubmission } from "../../../features/broadcast/hooks/useBroadcastSubmission";
+import { useBroadcast } from "../../../features/broadcast/hooks/useBroadcasts";
+
+export const Route = createFileRoute("/_dashboardLayout/broadcast/$id")({
+ head: () => ({
+ meta: [
+ {
+ title: "View Broadcast - OpenCircle Admin",
+ },
+ {
+ name: "description",
+ content: "View broadcast details",
+ },
+ ],
+ links: [
+ {
+ rel: "icon",
+ href: METADATA.favicon,
+ },
+ ],
+ }),
+ component: RouteComponent,
+});
+
+function RouteComponent() {
+ const { id } = Route.useParams();
+ const navigate = useNavigate();
+ const { data: broadcast, isLoading } = useBroadcast(id);
+ const { sendTestBroadcast, sendBroadcast, isSendingTest, isSending } =
+ useBroadcastSubmission();
+
+ const handleSendTest = async (testEmail: string) => {
+ sendTestBroadcast(
+ { id, testEmail },
+ {
+ onSuccess: () => {
+ alert("Test email queued successfully!");
+ },
+ onError: (error) => {
+ alert(`Failed to send test: ${error.message}`);
+ },
+ },
+ );
+ };
+
+ const handleSend = async () => {
+ if (
+ !window.confirm(
+ "Are you sure you want to send this broadcast to all users?",
+ )
+ ) {
+ return;
+ }
+
+ sendBroadcast(id, {
+ onSuccess: () => {
+ alert("Broadcast queued successfully!");
+ navigate({ to: "/broadcast" });
+ },
+ onError: (error) => {
+ alert(`Failed to send broadcast: ${error.message}`);
+ },
+ });
+ };
+
+ if (isLoading) {
+ return (
+
+
+
+ );
+ }
+
+ if (!broadcast) {
+ return (
+
+
+
Broadcast not found
+
+
+
+
+
+
+
+
+
Broadcasts
+
+
+
+
+
+
+ );
+}
diff --git a/apps/admin/src/routes/_dashboardLayout/broadcast/new.tsx b/apps/admin/src/routes/_dashboardLayout/broadcast/new.tsx
new file mode 100644
index 0000000..ab7da9b
--- /dev/null
+++ b/apps/admin/src/routes/_dashboardLayout/broadcast/new.tsx
@@ -0,0 +1,56 @@
+import { createFileRoute, useNavigate } from "@tanstack/react-router";
+import { METADATA } from "../../../constants/metadata";
+import { BroadcastEditor } from "../../../features/broadcast/components/broadcastEditor";
+import { useBroadcastSubmission } from "../../../features/broadcast/hooks/useBroadcastSubmission";
+import type {
+ BroadcastCreate,
+ BroadcastUpdate,
+} from "../../../features/broadcast/utils/types";
+
+export const Route = createFileRoute("/_dashboardLayout/broadcast/new")({
+ head: () => ({
+ meta: [
+ {
+ title: "Create Broadcast - OpenCircle Admin",
+ },
+ {
+ name: "description",
+ content: "Create a new broadcast email",
+ },
+ ],
+ links: [
+ {
+ rel: "icon",
+ href: METADATA.favicon,
+ },
+ ],
+ }),
+ component: RouteComponent,
+});
+
+function RouteComponent() {
+ const navigate = useNavigate();
+ const { createBroadcast, isCreating } = useBroadcastSubmission();
+
+ const handleSave = async (data: BroadcastCreate | BroadcastUpdate) => {
+ createBroadcast(data as BroadcastCreate, {
+ onSuccess: (broadcast) => {
+ navigate({ to: "/broadcast/$id", params: { id: broadcast.id } });
+ },
+ });
+ };
+
+ const handleCancel = () => {
+ navigate({ to: "/broadcast" });
+ };
+
+ return (
+
+
+
+ );
+}
diff --git a/apps/api/alembic/versions/1053d8244b76_add_channel_members_enum_value.py b/apps/api/alembic/versions/1053d8244b76_add_channel_members_enum_value.py
new file mode 100644
index 0000000..b174601
--- /dev/null
+++ b/apps/api/alembic/versions/1053d8244b76_add_channel_members_enum_value.py
@@ -0,0 +1,34 @@
+"""add channel_members enum value
+
+Revision ID: 1053d8244b76
+Revises: 232aa4832d70
+Create Date: 2025-11-28 23:09:16.186490
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+
+
+
+# revision identifiers, used by Alembic.
+revision: str = '1053d8244b76'
+down_revision: Union[str, Sequence[str], None] = '232aa4832d70'
+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.execute("ALTER TYPE broadcastrecipienttype ADD VALUE IF NOT EXISTS 'channel_members'")
+ # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+ """Downgrade schema."""
+ # ### commands auto generated by Alembic - please adjust! ###
+ pass
+ # ### end Alembic commands ###
diff --git a/apps/api/alembic/versions/232aa4832d70_update_broadcast_model.py b/apps/api/alembic/versions/232aa4832d70_update_broadcast_model.py
new file mode 100644
index 0000000..d1d0511
--- /dev/null
+++ b/apps/api/alembic/versions/232aa4832d70_update_broadcast_model.py
@@ -0,0 +1,36 @@
+"""update broadcast model
+
+Revision ID: 232aa4832d70
+Revises: 593648ef190a
+Create Date: 2025-11-28 23:02:18.949237
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+
+
+
+# revision identifiers, used by Alembic.
+revision: str = '232aa4832d70'
+down_revision: Union[str, Sequence[str], None] = '593648ef190a'
+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.add_column('broadcast', sa.Column('channel_id', sqlmodel.sql.sqltypes.AutoString(), nullable=True))
+ op.create_foreign_key(None, 'broadcast', 'channel', ['channel_id'], ['id'])
+ # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+ """Downgrade schema."""
+ # ### commands auto generated by Alembic - please adjust! ###
+ op.drop_constraint(None, 'broadcast', type_='foreignkey')
+ op.drop_column('broadcast', 'channel_id')
+ # ### end Alembic commands ###
diff --git a/apps/api/alembic/versions/2c9551fa3333_fix_enum_values_to_lowercase.py b/apps/api/alembic/versions/2c9551fa3333_fix_enum_values_to_lowercase.py
new file mode 100644
index 0000000..662e263
--- /dev/null
+++ b/apps/api/alembic/versions/2c9551fa3333_fix_enum_values_to_lowercase.py
@@ -0,0 +1,55 @@
+"""fix_enum_values_to_lowercase
+
+Revision ID: 2c9551fa3333
+Revises: 1053d8244b76
+Create Date: 2025-11-28 23:16:34.733476
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+
+
+
+# revision identifiers, used by Alembic.
+revision: str = '2c9551fa3333'
+down_revision: Union[str, Sequence[str], None] = '1053d8244b76'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+ """Upgrade schema."""
+ # Fix broadcastrecipienttype enum - rename UPPERCASE values to lowercase
+ # Note: 'channel_members' was already added in lowercase by previous migration
+ op.execute("ALTER TYPE broadcastrecipienttype RENAME VALUE 'ALL_USERS' TO 'all_users'")
+ op.execute("ALTER TYPE broadcastrecipienttype RENAME VALUE 'TEST_EMAIL' TO 'test_email'")
+
+ # Fix broadcaststatus enum - rename values to lowercase
+ op.execute("ALTER TYPE broadcaststatus RENAME VALUE 'DRAFT' TO 'draft'")
+ op.execute("ALTER TYPE broadcaststatus RENAME VALUE 'SENDING' TO 'sending'")
+ op.execute("ALTER TYPE broadcaststatus RENAME VALUE 'SENT' TO 'sent'")
+ op.execute("ALTER TYPE broadcaststatus RENAME VALUE 'FAILED' TO 'failed'")
+
+ # Fix broadcastrecipientstatus enum - rename values to lowercase
+ op.execute("ALTER TYPE broadcastrecipientstatus RENAME VALUE 'PENDING' TO 'pending'")
+ op.execute("ALTER TYPE broadcastrecipientstatus RENAME VALUE 'SENT' TO 'sent'")
+ op.execute("ALTER TYPE broadcastrecipientstatus RENAME VALUE 'FAILED' TO 'failed'")
+
+
+def downgrade() -> None:
+ """Downgrade schema."""
+ # Revert to uppercase values
+ op.execute("ALTER TYPE broadcastrecipienttype RENAME VALUE 'all_users' TO 'ALL_USERS'")
+ op.execute("ALTER TYPE broadcastrecipienttype RENAME VALUE 'test_email' TO 'TEST_EMAIL'")
+
+ op.execute("ALTER TYPE broadcaststatus RENAME VALUE 'draft' TO 'DRAFT'")
+ op.execute("ALTER TYPE broadcaststatus RENAME VALUE 'sending' TO 'SENDING'")
+ op.execute("ALTER TYPE broadcaststatus RENAME VALUE 'sent' TO 'SENT'")
+ op.execute("ALTER TYPE broadcaststatus RENAME VALUE 'failed' TO 'FAILED'")
+
+ op.execute("ALTER TYPE broadcastrecipientstatus RENAME VALUE 'pending' TO 'PENDING'")
+ op.execute("ALTER TYPE broadcastrecipientstatus RENAME VALUE 'sent' TO 'SENT'")
+ op.execute("ALTER TYPE broadcastrecipientstatus RENAME VALUE 'failed' TO 'FAILED'")
diff --git a/apps/api/alembic/versions/593648ef190a_add_broadcast_model.py b/apps/api/alembic/versions/593648ef190a_add_broadcast_model.py
new file mode 100644
index 0000000..dc10595
--- /dev/null
+++ b/apps/api/alembic/versions/593648ef190a_add_broadcast_model.py
@@ -0,0 +1,74 @@
+"""add broadcast model
+
+Revision ID: 593648ef190a
+Revises: f843c2a188ce
+Create Date: 2025-11-28 22:40:44.637736
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+
+
+
+# revision identifiers, used by Alembic.
+revision: str = '593648ef190a'
+down_revision: Union[str, Sequence[str], None] = 'f843c2a188ce'
+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('broadcast',
+ sa.Column('id', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+ sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+ sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
+ sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
+ sa.Column('subject', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+ sa.Column('content', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+ sa.Column('recipient_type', sa.Enum('ALL_USERS', 'TEST_EMAIL', name='broadcastrecipienttype'), nullable=False),
+ sa.Column('test_email', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+ sa.Column('status', sa.Enum('DRAFT', 'SENDING', 'SENT', 'FAILED', name='broadcaststatus'), nullable=False),
+ sa.Column('sent_at', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+ sa.Column('sent_count', sa.Integer(), nullable=False),
+ sa.Column('failed_count', sa.Integer(), nullable=False),
+ sa.Column('created_by', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+ sa.ForeignKeyConstraint(['created_by'], ['user.id'], ),
+ sa.PrimaryKeyConstraint('id')
+ )
+ op.create_index(op.f('ix_broadcast_subject'), 'broadcast', ['subject'], unique=False)
+ op.create_table('broadcast_recipient',
+ sa.Column('id', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+ sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+ sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
+ sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
+ sa.Column('broadcast_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+ sa.Column('user_id', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+ sa.Column('email', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+ sa.Column('status', sa.Enum('PENDING', 'SENT', 'FAILED', name='broadcastrecipientstatus'), nullable=False),
+ sa.Column('sent_at', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+ sa.Column('error_message', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+ sa.ForeignKeyConstraint(['broadcast_id'], ['broadcast.id'], ),
+ sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
+ sa.PrimaryKeyConstraint('id')
+ )
+ op.create_index(op.f('ix_broadcast_recipient_broadcast_id'), 'broadcast_recipient', ['broadcast_id'], unique=False)
+ op.create_index(op.f('ix_broadcast_recipient_email'), 'broadcast_recipient', ['email'], unique=False)
+ op.create_index(op.f('ix_broadcast_recipient_user_id'), 'broadcast_recipient', ['user_id'], unique=False)
+ # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+ """Downgrade schema."""
+ # ### commands auto generated by Alembic - please adjust! ###
+ op.drop_index(op.f('ix_broadcast_recipient_user_id'), table_name='broadcast_recipient')
+ op.drop_index(op.f('ix_broadcast_recipient_email'), table_name='broadcast_recipient')
+ op.drop_index(op.f('ix_broadcast_recipient_broadcast_id'), table_name='broadcast_recipient')
+ op.drop_table('broadcast_recipient')
+ op.drop_index(op.f('ix_broadcast_subject'), table_name='broadcast')
+ op.drop_table('broadcast')
+ # ### end Alembic commands ###
diff --git a/apps/api/pyproject.toml b/apps/api/pyproject.toml
index 88ca329..2d842cc 100644
--- a/apps/api/pyproject.toml
+++ b/apps/api/pyproject.toml
@@ -12,6 +12,7 @@ dependencies = [
"celery>=5.5.3",
"faker>=37.11.0",
"fastapi>=0.119.0",
+ "markdown>=3.7",
"passlib>=1.7.4",
"psycopg2-binary>=2.9.11",
"pydantic-settings>=2.11.0",
diff --git a/apps/api/src/api/broadcast/__init__.py b/apps/api/src/api/broadcast/__init__.py
new file mode 100644
index 0000000..8b13789
--- /dev/null
+++ b/apps/api/src/api/broadcast/__init__.py
@@ -0,0 +1 @@
+
diff --git a/apps/api/src/api/broadcast/api.py b/apps/api/src/api/broadcast/api.py
new file mode 100644
index 0000000..74f8c7f
--- /dev/null
+++ b/apps/api/src/api/broadcast/api.py
@@ -0,0 +1,203 @@
+from typing import List
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlmodel import Session
+
+from src.api.account.api import get_current_user
+from src.api.broadcast.serializer import (
+ BroadcastCreate,
+ BroadcastResponse,
+ BroadcastSendTest,
+ BroadcastUpdate,
+)
+from src.database.engine import get_session
+from src.database.models import BroadcastRecipientType, BroadcastStatus, Role, User
+from src.modules.broadcast.broadcast_methods import (
+ create_broadcast,
+ create_broadcast_recipients,
+ delete_broadcast,
+ get_all_active_users,
+ get_all_broadcasts,
+ get_broadcast,
+ get_channel_members_for_broadcast,
+ update_broadcast,
+ update_broadcast_status,
+)
+from src.modules.broadcast.broadcast_tasks import send_broadcast_task
+
+router = APIRouter()
+
+
+def require_admin(current_user: User = Depends(get_current_user)) -> User:
+ """Require admin role for access."""
+ if current_user.role != Role.ADMIN:
+ raise HTTPException(status_code=403, detail="Admin access required")
+ return current_user
+
+
+def build_broadcast_response(broadcast) -> dict:
+ """Build broadcast response data."""
+ return {
+ "id": broadcast.id,
+ "subject": broadcast.subject,
+ "content": broadcast.content,
+ "recipient_type": broadcast.recipient_type.value,
+ "test_email": broadcast.test_email,
+ "channel_id": broadcast.channel_id,
+ "channel": broadcast.channel,
+ "status": broadcast.status.value,
+ "sent_at": broadcast.sent_at,
+ "sent_count": broadcast.sent_count,
+ "failed_count": broadcast.failed_count,
+ "created_by": broadcast.created_by,
+ "creator": broadcast.creator,
+ "created_at": broadcast.created_at,
+ "updated_at": broadcast.updated_at,
+ }
+
+
+@router.post("/broadcasts/", response_model=BroadcastResponse)
+def create_broadcast_endpoint(
+ broadcast: BroadcastCreate,
+ db: Session = Depends(get_session),
+ current_user: User = Depends(require_admin),
+):
+ """Create a new broadcast."""
+ broadcast_data = broadcast.model_dump()
+ broadcast_data["created_by"] = current_user.id
+
+ if broadcast_data.get("recipient_type"):
+ broadcast_data["recipient_type"] = BroadcastRecipientType(
+ broadcast_data["recipient_type"]
+ )
+
+ created_broadcast = create_broadcast(db, broadcast_data)
+ full_broadcast = get_broadcast(db, created_broadcast.id)
+ return BroadcastResponse(**build_broadcast_response(full_broadcast))
+
+
+@router.get("/broadcasts/", response_model=List[BroadcastResponse])
+def get_all_broadcasts_endpoint(
+ skip: int = 0,
+ limit: int = 100,
+ db: Session = Depends(get_session),
+ current_user: User = Depends(require_admin),
+):
+ """Get all broadcasts."""
+ broadcasts = get_all_broadcasts(db, skip, limit)
+ return [BroadcastResponse(**build_broadcast_response(b)) for b in broadcasts]
+
+
+@router.get("/broadcasts/{broadcast_id}", response_model=BroadcastResponse)
+def get_broadcast_endpoint(
+ broadcast_id: str,
+ db: Session = Depends(get_session),
+ current_user: User = Depends(require_admin),
+):
+ """Get a broadcast by ID."""
+ broadcast = get_broadcast(db, broadcast_id)
+ if not broadcast:
+ raise HTTPException(status_code=404, detail="Broadcast not found")
+ return BroadcastResponse(**build_broadcast_response(broadcast))
+
+
+@router.put("/broadcasts/{broadcast_id}", response_model=BroadcastResponse)
+def update_broadcast_endpoint(
+ broadcast_id: str,
+ broadcast: BroadcastUpdate,
+ db: Session = Depends(get_session),
+ current_user: User = Depends(require_admin),
+):
+ """Update a broadcast."""
+ existing = get_broadcast(db, broadcast_id)
+ if not existing:
+ raise HTTPException(status_code=404, detail="Broadcast not found")
+
+ if existing.status != BroadcastStatus.DRAFT:
+ raise HTTPException(
+ status_code=400, detail="Can only update broadcasts in draft status"
+ )
+
+ update_data = {k: v for k, v in broadcast.model_dump().items() if v is not None}
+ if update_data.get("recipient_type"):
+ update_data["recipient_type"] = BroadcastRecipientType(
+ update_data["recipient_type"]
+ )
+
+ updated_broadcast = update_broadcast(db, broadcast_id, update_data)
+ return BroadcastResponse(**build_broadcast_response(updated_broadcast))
+
+
+@router.delete("/broadcasts/{broadcast_id}")
+def delete_broadcast_endpoint(
+ broadcast_id: str,
+ db: Session = Depends(get_session),
+ current_user: User = Depends(require_admin),
+):
+ """Delete a broadcast."""
+ if not delete_broadcast(db, broadcast_id):
+ raise HTTPException(status_code=404, detail="Broadcast not found")
+ return {"message": "Broadcast deleted"}
+
+
+@router.post("/broadcasts/{broadcast_id}/send-test")
+def send_test_broadcast_endpoint(
+ broadcast_id: str,
+ test_data: BroadcastSendTest,
+ db: Session = Depends(get_session),
+ current_user: User = Depends(require_admin),
+):
+ """Send a test broadcast to a specific email."""
+ broadcast = get_broadcast(db, broadcast_id)
+ if not broadcast:
+ raise HTTPException(status_code=404, detail="Broadcast not found")
+
+ create_broadcast_recipients(
+ db, broadcast_id, [{"email": test_data.test_email, "user_id": None}]
+ )
+
+ send_broadcast_task.delay(broadcast_id, is_test=True)
+
+ return {"message": f"Test broadcast queued for {test_data.test_email}"}
+
+
+@router.post("/broadcasts/{broadcast_id}/send")
+def send_broadcast_endpoint(
+ broadcast_id: str,
+ db: Session = Depends(get_session),
+ current_user: User = Depends(require_admin),
+):
+ """Send broadcast to users based on recipient type."""
+ broadcast = get_broadcast(db, broadcast_id)
+ if not broadcast:
+ raise HTTPException(status_code=404, detail="Broadcast not found")
+
+ if broadcast.status != BroadcastStatus.DRAFT:
+ raise HTTPException(
+ status_code=400, detail="Can only send broadcasts in draft status"
+ )
+
+ if broadcast.recipient_type == BroadcastRecipientType.CHANNEL_MEMBERS:
+ if not broadcast.channel_id:
+ raise HTTPException(
+ status_code=400,
+ detail="Channel ID required for channel members broadcast",
+ )
+ users = get_channel_members_for_broadcast(db, broadcast.channel_id)
+ if not users:
+ raise HTTPException(
+ status_code=400, detail="No active members in this channel"
+ )
+ else:
+ users = get_all_active_users(db)
+ if not users:
+ raise HTTPException(status_code=400, detail="No active users to send to")
+
+ recipients_data = [{"email": user.email, "user_id": user.id} for user in users]
+ create_broadcast_recipients(db, broadcast_id, recipients_data)
+
+ update_broadcast_status(db, broadcast_id, BroadcastStatus.SENDING)
+
+ send_broadcast_task.delay(broadcast_id, is_test=False)
+
+ return {"message": f"Broadcast queued for {len(users)} users"}
diff --git a/apps/api/src/api/broadcast/serializer.py b/apps/api/src/api/broadcast/serializer.py
new file mode 100644
index 0000000..5d0c52e
--- /dev/null
+++ b/apps/api/src/api/broadcast/serializer.py
@@ -0,0 +1,48 @@
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel
+
+from src.api.channels.serializer import ChannelResponse
+from src.api.user.serializer import UserResponse
+
+
+class BroadcastCreate(BaseModel):
+ subject: str
+ content: str
+ recipient_type: str = "test_email"
+ test_email: Optional[str] = None
+ channel_id: Optional[str] = None
+
+
+class BroadcastUpdate(BaseModel):
+ subject: Optional[str] = None
+ content: Optional[str] = None
+ recipient_type: Optional[str] = None
+ test_email: Optional[str] = None
+ channel_id: Optional[str] = None
+
+
+class BroadcastSendTest(BaseModel):
+ test_email: str
+
+
+class BroadcastResponse(BaseModel):
+ id: str
+ subject: str
+ content: str
+ recipient_type: str
+ test_email: Optional[str] = None
+ channel_id: Optional[str] = None
+ channel: Optional[ChannelResponse] = None
+ status: str
+ sent_at: Optional[str] = None
+ sent_count: int = 0
+ failed_count: int = 0
+ created_by: str
+ creator: Optional[UserResponse] = None
+ created_at: datetime
+ updated_at: datetime
+
+ class Config:
+ from_attributes = True
diff --git a/apps/api/src/api/channels/api.py b/apps/api/src/api/channels/api.py
index c41a995..32b5db6 100644
--- a/apps/api/src/api/channels/api.py
+++ b/apps/api/src/api/channels/api.py
@@ -7,7 +7,7 @@
from src.api.account.api import get_current_user
from src.api.resources.serializer import ResourceResponse
from src.database.engine import get_session as get_db
-from src.database.models import User
+from src.database.models import Role, User
from src.modules.channels.channels_methods import (
create_channel,
delete_channel,
@@ -77,11 +77,14 @@ def get_channel_endpoint(
# Check if user has access to this channel
# Public channels are accessible to everyone
- # Private channels require membership
+ # Private channels require membership (admins can access all)
if channel.type == "private":
from src.modules.channels.channels_methods import is_member
- if not current_user or not is_member(db, channel_id, current_user.id):
+ is_admin = current_user and current_user.role == Role.ADMIN
+ if not is_admin and (
+ not current_user or not is_member(db, channel_id, current_user.id)
+ ):
raise HTTPException(
status_code=403, detail="Access denied to private channel"
)
@@ -115,7 +118,8 @@ def update_channel_endpoint(
if existing_channel.type == "private":
from src.modules.channels.channels_methods import is_member
- if not is_member(db, channel_id, current_user.id):
+ is_admin = current_user.role == Role.ADMIN
+ if not is_admin and not is_member(db, channel_id, current_user.id):
raise HTTPException(
status_code=403, detail="Access denied to private channel"
)
@@ -141,7 +145,8 @@ def delete_channel_endpoint(
if existing_channel.type == "private":
from src.modules.channels.channels_methods import is_member
- if not is_member(db, channel_id, current_user.id):
+ is_admin = current_user.role == Role.ADMIN
+ if not is_admin and not is_member(db, channel_id, current_user.id):
raise HTTPException(
status_code=403, detail="Access denied to private channel"
)
@@ -166,7 +171,10 @@ def get_resources_by_channel_endpoint(
raise HTTPException(status_code=404, detail="Channel not found")
if channel.type == "private":
- if not current_user or not is_member(db, channel.id, current_user.id):
+ is_admin = current_user and current_user.role == Role.ADMIN
+ if not is_admin and (
+ not current_user or not is_member(db, channel.id, current_user.id)
+ ):
raise HTTPException(
status_code=403, detail="Access denied to private channel"
)
diff --git a/apps/api/src/core/celery_app.py b/apps/api/src/core/celery_app.py
index 7f4702a..175eaf7 100644
--- a/apps/api/src/core/celery_app.py
+++ b/apps/api/src/core/celery_app.py
@@ -9,6 +9,7 @@
backend=settings.CELERY_RESULT_BACKEND,
include=[
"src.modules.notifications.notification_tasks",
+ "src.modules.broadcast.broadcast_tasks",
],
)
diff --git a/apps/api/src/database/models.py b/apps/api/src/database/models.py
index bef2116..075c936 100644
--- a/apps/api/src/database/models.py
+++ b/apps/api/src/database/models.py
@@ -3,7 +3,8 @@
from enum import Enum
from typing import List, Optional
-from sqlalchemy import JSON
+from sqlalchemy import JSON, Column
+from sqlalchemy import Enum as SAEnum
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import Mapped, relationship
from sqlmodel import Field, Relationship
@@ -526,3 +527,83 @@ class PendingNotificationEmail(BaseModel, table=True):
notification: Mapped["Notification"] = Relationship(
sa_relationship=relationship("Notification")
)
+
+
+class BroadcastStatus(str, Enum):
+ DRAFT = "draft"
+ SENDING = "sending"
+ SENT = "sent"
+ FAILED = "failed"
+
+
+class BroadcastRecipientType(str, Enum):
+ ALL_USERS = "all_users"
+ TEST_EMAIL = "test_email"
+ CHANNEL_MEMBERS = "channel_members"
+
+
+class Broadcast(BaseModel, table=True):
+ __tablename__ = "broadcast" # type: ignore
+
+ subject: str = Field(index=True)
+ content: str # Markdown/HTML content from WYSIWYG
+ recipient_type: BroadcastRecipientType = Field(
+ default=BroadcastRecipientType.TEST_EMAIL,
+ sa_column=Column(
+ SAEnum(
+ BroadcastRecipientType, values_callable=lambda x: [e.value for e in x]
+ ),
+ default=BroadcastRecipientType.TEST_EMAIL,
+ ),
+ )
+ test_email: str | None = Field(default=None)
+ channel_id: str | None = Field(foreign_key="channel.id", default=None)
+ status: BroadcastStatus = Field(
+ default=BroadcastStatus.DRAFT,
+ sa_column=Column(
+ SAEnum(BroadcastStatus, values_callable=lambda x: [e.value for e in x]),
+ default=BroadcastStatus.DRAFT,
+ ),
+ )
+ sent_at: str | None = Field(default=None) # ISO datetime string
+ sent_count: int = Field(default=0)
+ failed_count: int = Field(default=0)
+ created_by: str = Field(foreign_key="user.id")
+ creator: Mapped["User"] = Relationship(
+ sa_relationship=relationship("User", foreign_keys="Broadcast.created_by")
+ )
+ channel: Mapped[Optional["Channel"]] = Relationship(
+ sa_relationship=relationship("Channel")
+ )
+ recipients: Mapped[List["BroadcastRecipient"]] = Relationship(
+ sa_relationship=relationship("BroadcastRecipient", back_populates="broadcast")
+ )
+
+
+class BroadcastRecipientStatus(str, Enum):
+ PENDING = "pending"
+ SENT = "sent"
+ FAILED = "failed"
+
+
+class BroadcastRecipient(BaseModel, table=True):
+ __tablename__ = "broadcast_recipient" # type: ignore
+
+ broadcast_id: str = Field(foreign_key="broadcast.id", index=True)
+ user_id: str | None = Field(foreign_key="user.id", default=None, index=True)
+ email: str = Field(index=True)
+ status: BroadcastRecipientStatus = Field(
+ default=BroadcastRecipientStatus.PENDING,
+ sa_column=Column(
+ SAEnum(
+ BroadcastRecipientStatus, values_callable=lambda x: [e.value for e in x]
+ ),
+ default=BroadcastRecipientStatus.PENDING,
+ ),
+ )
+ sent_at: str | None = Field(default=None) # ISO datetime string
+ error_message: str | None = Field(default=None)
+ broadcast: Mapped["Broadcast"] = Relationship(
+ sa_relationship=relationship("Broadcast", back_populates="recipients")
+ )
+ user: Mapped[Optional["User"]] = Relationship(sa_relationship=relationship("User"))
diff --git a/apps/api/src/main.py b/apps/api/src/main.py
index fa03b61..3f1d50f 100644
--- a/apps/api/src/main.py
+++ b/apps/api/src/main.py
@@ -10,6 +10,7 @@
from src.api.appsettings.api import router as appsettings_router
from src.api.article.api import router as article_router
from src.api.auth.api import router as auth_router
+from src.api.broadcast.api import router as broadcast_router
from src.api.channel_members.api import router as channel_members_router
from src.api.channels.api import router as channels_router
from src.api.courses.api import router as courses_router
@@ -80,6 +81,7 @@ async def lifespan(app: FastAPI):
app.include_router(auth_router, prefix="/api", tags=["auth"])
app.include_router(reaction_router, prefix="/api", tags=["reactions"])
app.include_router(article_router, prefix="/api", tags=["articles"])
+app.include_router(broadcast_router, prefix="/api", tags=["broadcasts"])
app.include_router(extras_router, prefix="/api", tags=["extras"])
app.include_router(invite_code_router, prefix="/api", tags=["invite-codes"])
app.include_router(notifications_router, prefix="/api", tags=["notifications"])
diff --git a/apps/api/src/modules/broadcast/__init__.py b/apps/api/src/modules/broadcast/__init__.py
new file mode 100644
index 0000000..8b13789
--- /dev/null
+++ b/apps/api/src/modules/broadcast/__init__.py
@@ -0,0 +1 @@
+
diff --git a/apps/api/src/modules/broadcast/broadcast_methods.py b/apps/api/src/modules/broadcast/broadcast_methods.py
new file mode 100644
index 0000000..eba4b81
--- /dev/null
+++ b/apps/api/src/modules/broadcast/broadcast_methods.py
@@ -0,0 +1,185 @@
+from datetime import datetime, timezone
+from typing import List, Optional
+
+from sqlalchemy import column, desc
+from sqlmodel import Session, select
+
+from src.database.models import (
+ Broadcast,
+ BroadcastRecipient,
+ BroadcastRecipientStatus,
+ BroadcastRecipientType,
+ BroadcastStatus,
+ ChannelMember,
+ User,
+)
+
+
+def create_broadcast(db: Session, broadcast_data: dict) -> Broadcast:
+ """Create a new broadcast."""
+ broadcast = Broadcast(
+ subject=broadcast_data["subject"],
+ content=broadcast_data["content"],
+ recipient_type=broadcast_data.get(
+ "recipient_type", BroadcastRecipientType.TEST_EMAIL
+ ),
+ test_email=broadcast_data.get("test_email"),
+ channel_id=broadcast_data.get("channel_id"),
+ status=BroadcastStatus.DRAFT,
+ created_by=broadcast_data["created_by"],
+ )
+ db.add(broadcast)
+ db.commit()
+ db.refresh(broadcast)
+ return broadcast
+
+
+def get_broadcast(db: Session, broadcast_id: str) -> Optional[Broadcast]:
+ """Get a broadcast by ID."""
+ statement = select(Broadcast).where(
+ Broadcast.id == broadcast_id, Broadcast.deleted_at.is_(None)
+ )
+ result = db.exec(statement).first()
+ return result
+
+
+def get_all_broadcasts(db: Session, skip: int = 0, limit: int = 100) -> List[Broadcast]:
+ """Get all broadcasts with pagination."""
+ statement = (
+ select(Broadcast)
+ .where(Broadcast.deleted_at.is_(None))
+ .order_by(desc(column("created_at")))
+ .offset(skip)
+ .limit(limit)
+ )
+ result = db.exec(statement).all()
+ return list(result)
+
+
+def update_broadcast(
+ db: Session, broadcast_id: str, update_data: dict
+) -> Optional[Broadcast]:
+ """Update a broadcast."""
+ broadcast = get_broadcast(db, broadcast_id)
+ if not broadcast:
+ return None
+
+ for key, value in update_data.items():
+ if value is not None:
+ setattr(broadcast, key, value)
+
+ db.commit()
+ db.refresh(broadcast)
+ return broadcast
+
+
+def delete_broadcast(db: Session, broadcast_id: str) -> bool:
+ """Soft delete a broadcast."""
+ broadcast = get_broadcast(db, broadcast_id)
+ if not broadcast:
+ return False
+
+ broadcast.deleted_at = datetime.now(timezone.utc)
+ db.commit()
+ return True
+
+
+def get_all_active_users(db: Session) -> List[User]:
+ """Get all active users for broadcast."""
+ statement = select(User).where(
+ User.is_active == True, # noqa: E712
+ User.deleted_at.is_(None),
+ )
+ result = db.exec(statement).all()
+ return list(result)
+
+
+def get_channel_members_for_broadcast(db: Session, channel_id: str) -> List[User]:
+ """Get all active users who are members of a channel."""
+ statement = (
+ select(User)
+ .join(ChannelMember, ChannelMember.user_id == User.id)
+ .where(
+ ChannelMember.channel_id == channel_id,
+ User.is_active == True, # noqa: E712
+ User.deleted_at.is_(None),
+ )
+ )
+ result = db.exec(statement).all()
+ return list(result)
+
+
+def create_broadcast_recipients(
+ db: Session, broadcast_id: str, emails: List[dict]
+) -> List[BroadcastRecipient]:
+ """Create broadcast recipients."""
+ recipients = []
+ for email_data in emails:
+ recipient = BroadcastRecipient(
+ broadcast_id=broadcast_id,
+ user_id=email_data.get("user_id"),
+ email=email_data["email"],
+ status=BroadcastRecipientStatus.PENDING,
+ )
+ db.add(recipient)
+ recipients.append(recipient)
+ db.commit()
+ for r in recipients:
+ db.refresh(r)
+ return recipients
+
+
+def update_broadcast_recipient_status(
+ db: Session,
+ recipient_id: str,
+ status: BroadcastRecipientStatus,
+ error_message: Optional[str] = None,
+) -> Optional[BroadcastRecipient]:
+ """Update broadcast recipient status."""
+ statement = select(BroadcastRecipient).where(BroadcastRecipient.id == recipient_id)
+ recipient = db.exec(statement).first()
+ if not recipient:
+ return None
+
+ recipient.status = status
+ if status == BroadcastRecipientStatus.SENT:
+ recipient.sent_at = datetime.now(timezone.utc).isoformat()
+ if error_message:
+ recipient.error_message = error_message
+
+ db.commit()
+ db.refresh(recipient)
+ return recipient
+
+
+def update_broadcast_status(
+ db: Session,
+ broadcast_id: str,
+ status: BroadcastStatus,
+ sent_count: int = 0,
+ failed_count: int = 0,
+) -> Optional[Broadcast]:
+ """Update broadcast status and counts."""
+ broadcast = get_broadcast(db, broadcast_id)
+ if not broadcast:
+ return None
+
+ broadcast.status = status
+ broadcast.sent_count = sent_count
+ broadcast.failed_count = failed_count
+ if status == BroadcastStatus.SENT:
+ broadcast.sent_at = datetime.now(timezone.utc).isoformat()
+
+ db.commit()
+ db.refresh(broadcast)
+ return broadcast
+
+
+def get_pending_recipients(db: Session, broadcast_id: str) -> List[BroadcastRecipient]:
+ """Get all pending recipients for a broadcast."""
+ statement = select(BroadcastRecipient).where(
+ BroadcastRecipient.broadcast_id == broadcast_id,
+ BroadcastRecipient.status == BroadcastRecipientStatus.PENDING,
+ )
+ result = db.exec(statement).all()
+ return list(result)
diff --git a/apps/api/src/modules/broadcast/broadcast_tasks.py b/apps/api/src/modules/broadcast/broadcast_tasks.py
new file mode 100644
index 0000000..3f36fd4
--- /dev/null
+++ b/apps/api/src/modules/broadcast/broadcast_tasks.py
@@ -0,0 +1,122 @@
+import re
+
+import markdown
+from loguru import logger
+from sqlmodel import Session
+
+from src.core.celery_app import celery_app
+from src.database.engine import engine
+from src.database.models import BroadcastRecipientStatus, BroadcastStatus
+from src.modules.broadcast.broadcast_methods import (
+ get_broadcast,
+ get_pending_recipients,
+ update_broadcast_recipient_status,
+ update_broadcast_status,
+)
+from src.modules.email.email_service import email_service
+
+EMAIL_REGEX = re.compile(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
+
+
+def markdown_to_html(content: str) -> str:
+ """Convert markdown content to HTML with email-friendly wrapper."""
+ html_body = markdown.markdown(
+ content,
+ extensions=["extra", "nl2br", "sane_lists"],
+ )
+ return f"""
+
+ {html_body}
+
+ """
+
+
+def is_valid_email(email: str) -> bool:
+ """Validate email format."""
+ if not email or not isinstance(email, str):
+ return False
+ return bool(EMAIL_REGEX.match(email.strip()))
+
+
+@celery_app.task
+def send_broadcast_task(broadcast_id: str, is_test: bool = False):
+ """Send broadcast emails as a background task."""
+ with Session(engine) as db:
+ try:
+ broadcast = get_broadcast(db, broadcast_id)
+ if not broadcast:
+ logger.error(f"Broadcast {broadcast_id} not found")
+ return {"success": False, "error": "Broadcast not found"}
+
+ pending_recipients = get_pending_recipients(db, broadcast_id)
+ if not pending_recipients:
+ logger.info(f"No pending recipients for broadcast {broadcast_id}")
+ return {"success": True, "message": "No pending recipients"}
+
+ sent_count = 0
+ failed_count = 0
+
+ for recipient in pending_recipients:
+ try:
+ if not is_valid_email(recipient.email):
+ logger.warning(f"Skipping invalid email: {recipient.email}")
+ update_broadcast_recipient_status(
+ db,
+ recipient.id,
+ BroadcastRecipientStatus.FAILED,
+ "Invalid email format",
+ )
+ failed_count += 1
+ continue
+
+ html_content = markdown_to_html(broadcast.content)
+ success = email_service.send_broadcast_email(
+ to_email=recipient.email.strip(),
+ subject=broadcast.subject,
+ html_content=html_content,
+ )
+
+ if success:
+ update_broadcast_recipient_status(
+ db, recipient.id, BroadcastRecipientStatus.SENT
+ )
+ sent_count += 1
+ else:
+ update_broadcast_recipient_status(
+ db,
+ recipient.id,
+ BroadcastRecipientStatus.FAILED,
+ "Failed to send email",
+ )
+ failed_count += 1
+
+ except Exception as e:
+ logger.error(f"Failed to send to {recipient.email}: {str(e)}")
+ update_broadcast_recipient_status(
+ db,
+ recipient.id,
+ BroadcastRecipientStatus.FAILED,
+ str(e),
+ )
+ failed_count += 1
+
+ if not is_test:
+ final_status = (
+ BroadcastStatus.SENT if sent_count > 0 else BroadcastStatus.FAILED
+ )
+ update_broadcast_status(
+ db, broadcast_id, final_status, sent_count, failed_count
+ )
+
+ return {
+ "success": True,
+ "sent_count": sent_count,
+ "failed_count": failed_count,
+ "message": f"Broadcast completed: {sent_count} sent, {failed_count} failed",
+ }
+
+ except Exception as e:
+ logger.error(f"Failed to process broadcast {broadcast_id}: {str(e)}")
+ if not is_test:
+ update_broadcast_status(db, broadcast_id, BroadcastStatus.FAILED, 0, 0)
+ return {"success": False, "error": str(e)}
diff --git a/apps/api/src/modules/email/email_service.py b/apps/api/src/modules/email/email_service.py
index 3b553af..0a759f6 100644
--- a/apps/api/src/modules/email/email_service.py
+++ b/apps/api/src/modules/email/email_service.py
@@ -201,6 +201,15 @@ def send_notification_digest_email(
return self._send_email(to_email, subject, html_content)
+ def send_broadcast_email(
+ self,
+ to_email: str,
+ subject: str,
+ html_content: str,
+ ) -> bool:
+ """Send broadcast email with custom subject and HTML content."""
+ return self._send_email(to_email, subject, html_content)
+
# Create a singleton instance
email_service = EmailService()
diff --git a/apps/api/uv.lock b/apps/api/uv.lock
index 6b9b92e..170dda9 100644
--- a/apps/api/uv.lock
+++ b/apps/api/uv.lock
@@ -66,6 +66,7 @@ dependencies = [
{ name = "fastapi" },
{ name = "httpx-oauth" },
{ name = "loguru" },
+ { name = "markdown" },
{ name = "passlib" },
{ name = "psycopg2-binary" },
{ name = "pydantic-settings" },
@@ -105,6 +106,7 @@ requires-dist = [
{ name = "fastapi", specifier = ">=0.119.0" },
{ name = "httpx-oauth", specifier = ">=0.15.0" },
{ name = "loguru", specifier = ">=0.7.3" },
+ { name = "markdown", specifier = ">=3.7" },
{ name = "passlib", specifier = ">=1.7.4" },
{ name = "psycopg2-binary", specifier = ">=2.9.11" },
{ name = "pydantic-settings", specifier = ">=2.11.0" },
@@ -672,6 +674,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/87/fb/99f81ac72ae23375f22b7afdb7642aba97c00a713c217124420147681a2f/mako-1.3.10-py3-none-any.whl", hash = "sha256:baef24a52fc4fc514a0887ac600f9f1cff3d82c61d4d700a1fa84d597b88db59", size = 78509, upload-time = "2025-04-10T12:50:53.297Z" },
]
+[[package]]
+name = "markdown"
+version = "3.10"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/7d/ab/7dd27d9d863b3376fcf23a5a13cb5d024aed1db46f963f1b5735ae43b3be/markdown-3.10.tar.gz", hash = "sha256:37062d4f2aa4b2b6b32aefb80faa300f82cc790cb949a35b8caede34f2b68c0e", size = 364931, upload-time = "2025-11-03T19:51:15.007Z" }
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/70/81/54e3ce63502cd085a0c556652a4e1b919c45a446bd1e5300e10c44c8c521/markdown-3.10-py3-none-any.whl", hash = "sha256:b5b99d6951e2e4948d939255596523444c0e677c669700b1d17aa4a8a464cb7c", size = 107678, upload-time = "2025-11-03T19:51:13.887Z" },
+]
+
[[package]]
name = "markdown-it-py"
version = "4.0.0"
diff --git a/apps/platform/package.json b/apps/platform/package.json
index 7b1b0a3..ddddbed 100644
--- a/apps/platform/package.json
+++ b/apps/platform/package.json
@@ -1,7 +1,7 @@
{
"name": "platform",
"private": true,
- "version": "0.0.14",
+ "version": "0.0.15",
"type": "module",
"scripts": {
"dev": "vite",
diff --git a/docs/www/package.json b/docs/www/package.json
index ba3c1d4..42bf639 100644
--- a/docs/www/package.json
+++ b/docs/www/package.json
@@ -1,6 +1,6 @@
{
"name": "www",
- "version": "0.0.14",
+ "version": "0.0.15",
"type": "module",
"private": true,
"scripts": {
diff --git a/package.json b/package.json
index fe5e951..71d0169 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
{
"name": "opencircle",
- "version": "0.0.14",
+ "version": "0.0.15",
"description": "",
"main": "index.js",
"scripts": {
diff --git a/packages/core/package.json b/packages/core/package.json
index 929f118..bd2e8d3 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -1,6 +1,6 @@
{
"name": "@opencircle/core",
- "version": "0.0.14",
+ "version": "0.0.15",
"description": "",
"main": "dist/index.js",
"types": "src/index.ts",
diff --git a/packages/core/src/services/index.ts b/packages/core/src/services/index.ts
index 588946c..17de18d 100644
--- a/packages/core/src/services/index.ts
+++ b/packages/core/src/services/index.ts
@@ -6,6 +6,7 @@ export {
} from "./routers/appsettings";
export { ArticlesRouter } from "./routers/articles";
export { AuthRouter } from "./routers/auth";
+export { BroadcastsRouter } from "./routers/broadcasts";
export { ChannelsRouter } from "./routers/channels";
export {
CoursesRouter,
diff --git a/packages/core/src/services/routers/broadcasts/index.ts b/packages/core/src/services/routers/broadcasts/index.ts
new file mode 100644
index 0000000..b1a532c
--- /dev/null
+++ b/packages/core/src/services/routers/broadcasts/index.ts
@@ -0,0 +1,52 @@
+import { BaseRouter } from "../../../utils/baseRouter";
+import type {
+ Broadcast,
+ BroadcastCreate,
+ BroadcastSendTest,
+ BroadcastUpdate,
+} from "./types";
+
+export class BroadcastsRouter extends BaseRouter {
+ async getAll(skip: number = 0, limit: number = 100): Promise {
+ const params = new URLSearchParams({
+ skip: skip.toString(),
+ limit: limit.toString(),
+ });
+ return this.client.get(`broadcasts/?${params.toString()}`);
+ }
+
+ async getById(broadcastId: string): Promise {
+ return this.client.get(`broadcasts/${broadcastId}`);
+ }
+
+ async create(data: BroadcastCreate): Promise {
+ return this.client.post("broadcasts/", data);
+ }
+
+ async update(broadcastId: string, data: BroadcastUpdate): Promise {
+ return this.client.put(`broadcasts/${broadcastId}`, data);
+ }
+
+ async delete(broadcastId: string): Promise<{ message: string }> {
+ return this.client.delete<{ message: string }>(`broadcasts/${broadcastId}`);
+ }
+
+ async sendTest(
+ broadcastId: string,
+ data: BroadcastSendTest,
+ ): Promise<{ message: string }> {
+ return this.client.post<{ message: string }>(
+ `broadcasts/${broadcastId}/send-test`,
+ data,
+ );
+ }
+
+ async send(broadcastId: string): Promise<{ message: string }> {
+ return this.client.post<{ message: string }>(
+ `broadcasts/${broadcastId}/send`,
+ {},
+ );
+ }
+}
+
+export * from "./types";
diff --git a/packages/core/src/services/routers/broadcasts/types.ts b/packages/core/src/services/routers/broadcasts/types.ts
new file mode 100644
index 0000000..89088cc
--- /dev/null
+++ b/packages/core/src/services/routers/broadcasts/types.ts
@@ -0,0 +1,46 @@
+import type { User } from "../auth/types";
+import type { Channel } from "../channels/types";
+
+export type BroadcastStatus = "draft" | "sending" | "sent" | "failed";
+export type BroadcastRecipientType =
+ | "all_users"
+ | "test_email"
+ | "channel_members";
+
+export interface Broadcast {
+ id: string;
+ subject: string;
+ content: string;
+ recipient_type: BroadcastRecipientType;
+ test_email?: string;
+ channel_id?: string;
+ channel?: Channel;
+ status: BroadcastStatus;
+ sent_at?: string;
+ sent_count: number;
+ failed_count: number;
+ created_by: string;
+ creator?: User;
+ created_at: string;
+ updated_at: string;
+}
+
+export interface BroadcastCreate {
+ subject: string;
+ content: string;
+ recipient_type?: BroadcastRecipientType;
+ test_email?: string;
+ channel_id?: string;
+}
+
+export interface BroadcastUpdate {
+ subject?: string;
+ content?: string;
+ recipient_type?: BroadcastRecipientType;
+ test_email?: string;
+ channel_id?: string;
+}
+
+export interface BroadcastSendTest {
+ test_email: string;
+}
diff --git a/packages/core/src/services/types.ts b/packages/core/src/services/types.ts
index 5723eed..7461b8b 100644
--- a/packages/core/src/services/types.ts
+++ b/packages/core/src/services/types.ts
@@ -4,6 +4,7 @@ export * from "./routers/applinks/types";
export * from "./routers/appsettings/types";
export * from "./routers/articles/types";
export * from "./routers/auth/types";
+export * from "./routers/broadcasts/types";
export * from "./routers/channels/types";
export * from "./routers/courses/types";
export * from "./routers/extras/types";
diff --git a/packages/core/src/utils/api.ts b/packages/core/src/utils/api.ts
index c20c577..f2b34b2 100644
--- a/packages/core/src/utils/api.ts
+++ b/packages/core/src/utils/api.ts
@@ -4,6 +4,7 @@ import { AppLinksRouter } from "../services/routers/applinks";
import { AppSettingsRouter } from "../services/routers/appsettings";
import { ArticlesRouter } from "../services/routers/articles";
import { AuthRouter } from "../services/routers/auth";
+import { BroadcastsRouter } from "../services/routers/broadcasts";
import { ChannelsRouter } from "../services/routers/channels";
import { CoursesRouter } from "../services/routers/courses";
import { ExtrasRouter } from "../services/routers/extras";
@@ -21,6 +22,7 @@ export class Api {
public auth: AuthRouter;
public appSettings: AppSettingsRouter;
public appLinks: AppLinksRouter;
+ public broadcasts: BroadcastsRouter;
public channels: ChannelsRouter;
public posts: PostsRouter;
public media: MediaRouter;
@@ -39,6 +41,7 @@ export class Api {
this.auth = new AuthRouter(baseUrl, hooks);
this.appSettings = new AppSettingsRouter(baseUrl, hooks);
this.appLinks = new AppLinksRouter(baseUrl, hooks);
+ this.broadcasts = new BroadcastsRouter(baseUrl, hooks);
this.channels = new ChannelsRouter(baseUrl, hooks);
this.posts = new PostsRouter(baseUrl, hooks);
this.media = new MediaRouter(baseUrl, hooks);
diff --git a/packages/ui/package.json b/packages/ui/package.json
index b70e2d6..4fb4da2 100644
--- a/packages/ui/package.json
+++ b/packages/ui/package.json
@@ -1,6 +1,6 @@
{
"name": "@opencircle/ui",
- "version": "0.0.14",
+ "version": "0.0.15",
"description": "",
"main": "src/index.ts",
"scripts": {