feat: add share calendar API

backend/src/v1/schedule_items/repository.py

@@ -10,6 +10,7 @@ from sqlalchemy.exc import SQLAlchemyError
 from core.db.base_repository import BaseRepository
 from core.logging import get_logger
 from models.schedule_items import ScheduleItem
+from models.schedule_subscriptions import ScheduleSubscription
 
 if TYPE_CHECKING:
     from sqlalchemy.ext.asyncio import AsyncSession
@@ -18,6 +19,7 @@ logger = get_logger("v1.schedule_items.repository")
 
 
 class ScheduleItemRepository(Protocol):
+    async def get_by_id(self, entity_id: UUID) -> ScheduleItem | None: ...
     async def get_by_item_id(
         self, item_id: UUID, owner_id: UUID
     ) -> ScheduleItem | None: ...
@@ -31,6 +33,7 @@ class ScheduleItemRepository(Protocol):
     async def list_by_date_range(
         self, owner_id: UUID, start_at: datetime, end_at: datetime
     ) -> list[ScheduleItem]: ...
+    async def create_subscription(self, data: dict) -> ScheduleSubscription: ...
 
 
 class SQLAlchemyScheduleItemRepository(BaseRepository[ScheduleItem]):
@@ -127,3 +130,12 @@ class SQLAlchemyScheduleItemRepository(BaseRepository[ScheduleItem]):
         except SQLAlchemyError:
             logger.exception("Schedule item list failed", owner_id=str(owner_id))
             raise
+
+    async def create_subscription(self, data: dict) -> ScheduleSubscription:
+        sub = ScheduleSubscription(**data)
+        self._session.add(sub)
+        await self._session.flush()
+        return sub
+
+    async def get_by_id(self, entity_id: UUID) -> ScheduleItem | None:
+        return await super().get_by_id(entity_id)

backend/src/v1/schedule_items/router.py

@@ -12,6 +12,8 @@ from v1.schedule_items.schemas import (
     ScheduleItemListItem,
     ScheduleItemListRequest,
     ScheduleItemResponse,
+    ScheduleItemShareRequest,
+    ScheduleItemShareResponse,
     ScheduleItemUpdateRequest,
 )
 from v1.schedule_items.service import ScheduleItemService
@@ -72,3 +74,12 @@ async def delete_schedule_item(
     service: Annotated[ScheduleItemService, Depends(get_schedule_item_service)],
 ) -> None:
     await service.delete(item_id)
+
+
+@router.post("/{item_id}/share", response_model=ScheduleItemShareResponse)
+async def share_schedule_item(
+    item_id: UUID,
+    request: ScheduleItemShareRequest,
+    service: Annotated[ScheduleItemService, Depends(get_schedule_item_service)],
+) -> ScheduleItemShareResponse:
+    return await service.share(item_id, request)

backend/src/v1/schedule_items/schemas.py

@@ -96,3 +96,24 @@ class ScheduleItemListItem(BaseModel):
 class ScheduleItemListRequest(BaseModel):
     start_at: datetime
     end_at: datetime
+
+
+class ScheduleItemShareRequest(BaseModel):
+    email: str = Field(..., description="Email of user to share with")
+    permission_view: bool = Field(True, description="Grant view permission")
+    permission_edit: bool = Field(False, description="Grant edit permission")
+    permission_invite: bool = Field(False, description="Grant invite permission")
+
+    def _permission_value(self) -> int:
+        value = 0
+        if self.permission_view:
+            value |= 1
+        if self.permission_edit:
+            value |= 4
+        if self.permission_invite:
+            value |= 2
+        return value
+
+
+class ScheduleItemShareResponse(BaseModel):
+    message: str

backend/src/v1/schedule_items/service.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+import json
+from typing import TYPE_CHECKING, Protocol
 from uuid import UUID
 
 from fastapi import HTTPException
@@ -9,13 +10,17 @@ from sqlalchemy.exc import SQLAlchemyError
 from core.auth.models import CurrentUser
 from core.db.base_service import BaseService
 from core.logging import get_logger
+from models.inbox_messages import InboxMessage, InboxMessageType
 from models.schedule_items import ScheduleItem
+from v1.auth.gateway import SupabaseAuthGateway
 from v1.schedule_items.repository import ScheduleItemRepository
 from v1.schedule_items.schemas import (
     ScheduleItemCreateRequest,
     ScheduleItemListRequest,
     ScheduleItemMetadata,
     ScheduleItemResponse,
+    ScheduleItemShareRequest,
+    ScheduleItemShareResponse,
     ScheduleItemUpdateRequest,
     ScheduleItemSourceType,
     ScheduleItemStatus,
@@ -24,22 +29,31 @@ from v1.schedule_items.schemas import (
 if TYPE_CHECKING:
     from sqlalchemy.ext.asyncio import AsyncSession
 
+    from v1.auth.schemas import UserByEmailResponse
+
 logger = get_logger("v1.schedule_items.service")
 
 
+class AuthByEmailGateway(Protocol):
+    async def get_user_by_email(self, email: str) -> "UserByEmailResponse": ...
+
+
 class ScheduleItemService(BaseService):
     _repository: ScheduleItemRepository
     _session: AsyncSession
+    _auth_gateway: AuthByEmailGateway
 
     def __init__(
         self,
         repository: ScheduleItemRepository,
         session: AsyncSession,
         current_user: CurrentUser | None,
+        auth_gateway: AuthByEmailGateway | None = None,
     ) -> None:
         super().__init__(current_user=current_user)
         self._repository = repository
         self._session = session
+        self._auth_gateway = auth_gateway or SupabaseAuthGateway()
 
     async def create(self, request: ScheduleItemCreateRequest) -> ScheduleItemResponse:
         user_id = self.require_user_id()
@@ -179,6 +193,50 @@ class ScheduleItemService(BaseService):
 
         return [self._to_response(item) for item in items]
 
+    async def share(
+        self, item_id: UUID, request: ScheduleItemShareRequest
+    ) -> ScheduleItemShareResponse:
+        user_id = self.require_user_id()
+
+        try:
+            item = await self._repository.get_by_id(item_id)
+            if item is None:
+                raise HTTPException(status_code=404, detail="Schedule item not found")
+            if item.owner_id != user_id:
+                raise HTTPException(
+                    status_code=403,
+                    detail="Only owner can share this schedule item",
+                )
+
+            target_user = await self._auth_gateway.get_user_by_email(request.email)
+            recipient_id = UUID(target_user.id)
+            message = InboxMessage(
+                recipient_id=recipient_id,
+                sender_id=user_id,
+                message_type=InboxMessageType.CALENDAR,
+                schedule_item_id=item.id,
+                content=json.dumps({"permission": request._permission_value()}),
+                created_by=user_id,
+            )
+            self._session.add(message)
+            await self._session.commit()
+        except HTTPException:
+            raise
+        except SQLAlchemyError:
+            await self._session.rollback()
+            logger.exception("Failed to share schedule item", item_id=str(item_id))
+            raise HTTPException(
+                status_code=503, detail="Schedule item store unavailable"
+            )
+        except ValueError:
+            await self._session.rollback()
+            logger.exception(
+                "Auth lookup returned invalid user id", email=request.email
+            )
+            raise HTTPException(status_code=503, detail="Auth lookup unavailable")
+
+        return ScheduleItemShareResponse(message="Calendar invitation sent")
+
     def _to_response(self, item: ScheduleItem) -> ScheduleItemResponse:
         return ScheduleItemResponse(
             id=item.id,

backend/tests/unit/v1/schedule_items/test_share.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import cast
+from unittest.mock import AsyncMock, MagicMock
+from uuid import UUID, uuid4
+
+import pytest
+from fastapi import HTTPException
+
+from core.auth.models import CurrentUser
+from models.schedule_items import ScheduleItem
+from v1.auth.schemas import UserByEmailResponse
+from v1.schedule_items.repository import ScheduleItemRepository
+from v1.schedule_items.schemas import ScheduleItemShareRequest
+from v1.schedule_items.service import ScheduleItemService
+
+
+def test_share_request_schema() -> None:
+    request = ScheduleItemShareRequest(
+        email="friend@example.com",
+        permission_view=True,
+        permission_edit=True,
+        permission_invite=False,
+    )
+    assert request.email == "friend@example.com"
+    assert request.permission_view is True
+
+
+def test_permission_bits_calculation() -> None:
+    request = ScheduleItemShareRequest(
+        email="friend@example.com",
+        permission_view=True,
+        permission_edit=True,
+        permission_invite=False,
+    )
+    assert request._permission_value() == 5
+
+
+def _build_item(item_id: UUID, owner_id: UUID) -> ScheduleItem:
+    item = MagicMock(spec=ScheduleItem)
+    item.id = item_id
+    item.owner_id = owner_id
+    item.title = "test"
+    item.description = None
+    item.start_at = datetime(2026, 2, 28, 16, 0, 0, tzinfo=timezone.utc)
+    item.end_at = None
+    item.timezone = "UTC"
+    item.extra_metadata = {}
+    item.created_at = datetime(2026, 2, 28, 10, 0, 0, tzinfo=timezone.utc)
+    item.updated_at = datetime(2026, 2, 28, 10, 0, 0, tzinfo=timezone.utc)
+    return item
+
+
+class ShareRepo:
+    def __init__(self, item: ScheduleItem | None) -> None:
+        self._item = item
+
+    async def get_by_id(self, item_id: UUID) -> ScheduleItem | None:
+        if self._item and self._item.id == item_id:
+            return self._item
+        return None
+
+
+class AuthGatewayStub:
+    async def get_user_by_email(self, email: str) -> UserByEmailResponse:
+        return UserByEmailResponse(
+            id="00000000-0000-0000-0000-000000000222",
+            email=email,
+            created_at="2026-02-28T10:00:00Z",
+            email_confirmed_at=None,
+        )
+
+
+@pytest.mark.asyncio
+async def test_share_forbidden_when_not_owner() -> None:
+    owner_id = UUID("00000000-0000-0000-0000-000000000001")
+    requester_id = UUID("00000000-0000-0000-0000-000000000002")
+    item_id = uuid4()
+    service = ScheduleItemService(
+        repository=cast(
+            ScheduleItemRepository,
+            ShareRepo(_build_item(item_id=item_id, owner_id=owner_id)),
+        ),
+        session=AsyncMock(),
+        current_user=CurrentUser(id=requester_id),
+        auth_gateway=AuthGatewayStub(),
+    )
+
+    with pytest.raises(HTTPException) as exc_info:
+        await service.share(
+            item_id,
+            ScheduleItemShareRequest(
+                email="friend@example.com",
+                permission_view=True,
+                permission_edit=False,
+                permission_invite=False,
+            ),
+        )
+
+    assert exc_info.value.status_code == 403