Google Calendar

<?php

declare(strict_types=1);

final readonly class CreateEventRequest
{
    /**
     * @param list<string> $attendees   // emails
     * @param list<int>    $remindersMinutes
     */
    public function __construct(
        public string $title,
        public \DateTimeImmutable $startsAt,      // in UTC
        public \DateTimeImmutable $endsAt,        // in UTC
        public string $timezone,                  // IANA id, e.g. "Europe/Moscow"
        public ?string $location = null,
        public ?string $description = null,
        public ?string $rrule = null,             // e.g. "FREQ=WEEKLY;BYDAY=TU;UNTIL=20260901T000000Z"
        public array $attendees = [],
        public array $remindersMinutes = [10],
    ) {}
}

final readonly class EventDTO
{
    public function __construct(
        public string $id,
        public string $title,
        public \DateTimeImmutable $startsAt,
        public \DateTimeImmutable $endsAt,
        public string $timezone,
        public ?string $rrule,
        public ?string $recurrenceId,   // если это override серии
        public ?string $masterId,        // ссылка на родителя (для overrides)
    ) {}
}

package api

import "time"

type CreateEventRequest struct {
    Title            string    `json:"title"`
    StartsAt         time.Time `json:"starts_at"`
    EndsAt           time.Time `json:"ends_at"`
    Timezone         string    `json:"timezone"`
    RRule            string    `json:"rrule,omitempty"`
    Attendees        []string  `json:"attendees,omitempty"`
    RemindersMinutes []int     `json:"reminders_minutes,omitempty"`
}

type FreeBusyRequest struct {
    UserIDs []string  `json:"users"`
    From    time.Time `json:"from"`
    To      time.Time `json:"to"`
}

type Interval struct {
    Start time.Time `json:"start"`
    End   time.Time `json:"end"`
}

type FreeBusyResponse struct {
    Busy map[string][]Interval `json:"busy"`
}

namespace Calendar.Api;

// DateTimeOffset keeps the absolute instant; Timezone carries the IANA id separately.
public sealed record CreateEventRequest(
    string Title,
    DateTimeOffset StartsAt,                     // in UTC
    DateTimeOffset EndsAt,                       // in UTC
    string Timezone,                             // IANA id, e.g. "Europe/Moscow"
    string? Location = null,
    string? Description = null,
    string? RRule = null,                        // e.g. "FREQ=WEEKLY;BYDAY=TU;UNTIL=20260901T000000Z"
    IReadOnlyList<string>? Attendees = null,     // emails
    IReadOnlyList<int>? RemindersMinutes = null);

public sealed record EventDto(
    Guid Id,
    string Title,
    DateTimeOffset StartsAt,
    DateTimeOffset EndsAt,
    string Timezone,
    string? RRule,
    DateTimeOffset? RecurrenceId,   // set when this is a series override
    Guid? MasterId);                // parent series (for overrides)

public sealed record Interval(DateTimeOffset Start, DateTimeOffset End);

public sealed record FreeBusyRequest(
    IReadOnlyList<Guid> UserIds,
    DateTimeOffset From,
    DateTimeOffset To);

public sealed record FreeBusyResponse(
    IReadOnlyDictionary<Guid, IReadOnlyList<Interval>> Busy);

from dataclasses import dataclass, field
from datetime import datetime
from uuid import UUID

# All datetimes are timezone-aware and stored in UTC; timezone holds the IANA id.

@dataclass(frozen=True, slots=True)
class CreateEventRequest:
    title: str
    starts_at: datetime                       # in UTC
    ends_at: datetime                         # in UTC
    timezone: str                             # IANA id, e.g. "Europe/Moscow"
    location: str | None = None
    description: str | None = None
    rrule: str | None = None                  # e.g. "FREQ=WEEKLY;BYDAY=TU;UNTIL=20260901T000000Z"
    attendees: list[str] = field(default_factory=list)          # emails
    reminders_minutes: list[int] = field(default_factory=lambda: [10])

@dataclass(frozen=True, slots=True)
class EventDTO:
    id: UUID
    title: str
    starts_at: datetime
    ends_at: datetime
    timezone: str
    rrule: str | None
    recurrence_id: datetime | None   # set when this is a series override
    master_id: UUID | None           # parent series (for overrides)

@dataclass(frozen=True, slots=True)
class Interval:
    start: datetime
    end: datetime

@dataclass(frozen=True, slots=True)
class FreeBusyRequest:
    user_ids: list[UUID]
    from_: datetime   # "from" is a keyword, hence the trailing underscore
    to: datetime

@dataclass(frozen=True, slots=True)
class FreeBusyResponse:
    busy: dict[UUID, list[Interval]]

<?php

declare(strict_types=1);

use Recurr\Rule;
use Recurr\Transformer\ArrayTransformer;
use Recurr\Transformer\Constraint\BetweenConstraint;

final class RruleExpander
{
    public function expand(
        \DateTimeImmutable $masterStart,
        string $rrule,
        string $timezone,
        \DateTimeImmutable $from,
        \DateTimeImmutable $to,
    ): array {
        $tz = new \DateTimeZone($timezone);
        // Recurr operates on DateTime instances in a specific timezone.
        $startInTz = $masterStart->setTimezone($tz);

        $rule = new Rule($rrule, \DateTime::createFromImmutable($startInTz), null, $timezone);
        $transformer = new ArrayTransformer();

        $constraint = new BetweenConstraint(
            \DateTime::createFromImmutable($from),
            \DateTime::createFromImmutable($to),
            true,
        );

        $occurrences = $transformer->transform($rule, $constraint);

        $results = [];
        foreach ($occurrences as $o) {
            $results[] = \DateTimeImmutable::createFromMutable($o->getStart());
        }
        return $results;
    }
}

package rrule

import (
    "time"

    rr "github.com/teambition/rrule-go"
)

// Expand returns occurrences of rrule between from and to (UTC).
func Expand(masterStart time.Time, rruleStr, tz string, from, to time.Time) ([]time.Time, error) {
    loc, err := time.LoadLocation(tz)
    if err != nil {
        return nil, err
    }
    startLocal := masterStart.In(loc)

    opt, err := rr.StrToROption(rruleStr)
    if err != nil {
        return nil, err
    }
    opt.Dtstart = startLocal

    r, err := rr.NewRRule(*opt)
    if err != nil {
        return nil, err
    }
    return r.Between(from, to, true), nil
}

using Ical.Net.CalendarComponents;
using Ical.Net.DataTypes;

namespace Calendar.Recurrence;

public sealed class RRuleExpander
{
    // Expands rrule into occurrences between from and to, returned in UTC.
    public IReadOnlyList<DateTimeOffset> Expand(
        DateTimeOffset masterStart,
        string rrule,
        string timezone,
        DateTimeOffset from,
        DateTimeOffset to)
    {
        // .NET 6+ resolves IANA ids on every platform, including Windows.
        var tz = TimeZoneInfo.FindSystemTimeZoneById(timezone);
        var startLocal = TimeZoneInfo.ConvertTime(masterStart, tz);

        var evt = new CalendarEvent
        {
            Start = new CalDateTime(startLocal.DateTime, timezone),
            Duration = TimeSpan.Zero,
            RecurrenceRules = { new RecurrencePattern(rrule) },
        };

        return evt.GetOccurrences(from.UtcDateTime, to.UtcDateTime)
            .Select(o => new DateTimeOffset(o.Period.StartTime.AsUtc, TimeSpan.Zero))
            .OrderBy(d => d)
            .ToList();
    }
}

from datetime import datetime, timezone as utc_tz
from zoneinfo import ZoneInfo

from dateutil.rrule import rrulestr

def expand(
    master_start: datetime,
    rrule: str,
    tz_name: str,
    from_: datetime,
    to: datetime,
) -> list[datetime]:
    """Return occurrences of rrule between from_ and to (UTC)."""
    tz = ZoneInfo(tz_name)
    # dateutil expands in the tz of dtstart, so DST shifts keep the wall-clock time.
    start_local = master_start.astimezone(tz)

    rule = rrulestr(rrule, dtstart=start_local)

    # Aware datetimes compare across zones, so from_/to may stay in UTC.
    return [o.astimezone(utc_tz) for o in rule.between(from_, to, inc=True)]

<?php

declare(strict_types=1);

final class ReminderWorker
{
    public function __construct(
        private readonly \PDO $db,
        private readonly Notifier $notifier,
        private readonly LoggerInterface $logger,
    ) {}

    public function runBatch(int $shardId, int $shardCount): int
    {
        $stmt = $this->db->prepare(<<<SQL
            SELECT id, event_id, user_id, fire_at
            FROM reminders
            WHERE fire_at <= now()
              AND NOT delivered
              AND hashtext(user_id::text) % :shards = :shard
            ORDER BY fire_at
            LIMIT 5000
            FOR UPDATE SKIP LOCKED
        SQL);
        $stmt->execute(['shards' => $shardCount, 'shard' => $shardId]);
        $rows = $stmt->fetchAll(\PDO::FETCH_ASSOC);

        $count = 0;
        foreach ($rows as $r) {
            try {
                $this->notifier->send($r['user_id'], $r['event_id']);
                $this->db->prepare('UPDATE reminders SET delivered=true WHERE id = :id')
                    ->execute(['id' => $r['id']]);
                $count++;
            } catch (\Throwable $e) {
                $this->logger->error('reminder failed', ['id' => $r['id'], 'err' => $e->getMessage()]);
            }
        }
        return $count;
    }
}

using Microsoft.Extensions.Logging;
using Npgsql;

namespace Calendar.Reminders;

public sealed class ReminderWorker(
    NpgsqlDataSource dataSource,
    INotifier notifier,
    ILogger<ReminderWorker> logger)
{
    private const string SelectBatch = """
        SELECT id, event_id, user_id, fire_at
        FROM reminders
        WHERE fire_at <= now()
          AND NOT delivered
          AND hashtext(user_id::text) % @shards = @shard
        ORDER BY fire_at
        LIMIT 5000
        FOR UPDATE SKIP LOCKED
        """;

    public async Task<int> RunBatchAsync(int shardId, int shardCount, CancellationToken ct)
    {
        await using var conn = await dataSource.OpenConnectionAsync(ct);
        // SKIP LOCKED holds the row locks until commit, so the batch runs in one transaction.
        await using var tx = await conn.BeginTransactionAsync(ct);

        var rows = new List<(Guid Id, Guid EventId, Guid UserId)>();
        await using (var cmd = new NpgsqlCommand(SelectBatch, conn, tx))
        {
            cmd.Parameters.AddWithValue("shards", shardCount);
            cmd.Parameters.AddWithValue("shard", shardId);
            await using var reader = await cmd.ExecuteReaderAsync(ct);
            while (await reader.ReadAsync(ct))
            {
                rows.Add((reader.GetGuid(0), reader.GetGuid(1), reader.GetGuid(2)));
            }
        }

        var count = 0;
        foreach (var (id, eventId, userId) in rows)
        {
            try
            {
                await notifier.SendAsync(userId, eventId, ct);
                await using var upd = new NpgsqlCommand(
                    "UPDATE reminders SET delivered=true WHERE id = @id", conn, tx);
                upd.Parameters.AddWithValue("id", id);
                await upd.ExecuteNonQueryAsync(ct);
                count++;
            }
            catch (Exception e) when (e is not OperationCanceledException)
            {
                logger.LogError(e, "reminder failed {ReminderId}", id);
            }
        }

        await tx.CommitAsync(ct);
        return count;
    }
}

import logging

from psycopg_pool import AsyncConnectionPool

logger = logging.getLogger(__name__)

SELECT_BATCH = """
    SELECT id, event_id, user_id, fire_at
    FROM reminders
    WHERE fire_at <= now()
      AND NOT delivered
      AND hashtext(user_id::text) %% %(shards)s = %(shard)s
    ORDER BY fire_at
    LIMIT 5000
    FOR UPDATE SKIP LOCKED
"""

class ReminderWorker:
    def __init__(self, pool: AsyncConnectionPool, notifier: "Notifier") -> None:
        self._pool = pool
        self._notifier = notifier

    async def run_batch(self, shard_id: int, shard_count: int) -> int:
        # SKIP LOCKED holds the row locks until commit, so the batch runs in one transaction.
        async with self._pool.connection() as conn, conn.transaction():
            async with conn.cursor() as cur:
                await cur.execute(SELECT_BATCH, {"shards": shard_count, "shard": shard_id})
                rows = await cur.fetchall()

                count = 0
                for reminder_id, event_id, user_id, _fire_at in rows:
                    try:
                        await self._notifier.send(user_id, event_id)
                        await cur.execute(
                            "UPDATE reminders SET delivered=true WHERE id = %s",
                            (reminder_id,),
                        )
                        count += 1
                    except Exception:
                        logger.exception("reminder failed", extra={"id": str(reminder_id)})
                return count

package freebusy

import (
    "context"
    "database/sql"
    "sort"
    "time"
)

type Interval struct {
    Start time.Time
    End   time.Time
}

// FindSlots returns available slots of given duration that are free for all users.
func FindSlots(ctx context.Context, db *sql.DB, users []string, from, to time.Time, duration time.Duration) ([]Interval, error) {
    busy := make([]Interval, 0)
    for _, u := range users {
        rows, err := db.QueryContext(ctx,
            `SELECT starts_at, ends_at FROM free_busy_intervals
             WHERE user_id = $1 AND ends_at > $2 AND starts_at < $3`,
            u, from, to)
        if err != nil {
            return nil, err
        }
        for rows.Next() {
            var iv Interval
            if err := rows.Scan(&iv.Start, &iv.End); err != nil {
                rows.Close()
                return nil, err
            }
            busy = append(busy, iv)
        }
        rows.Close()
    }

    // Merge overlapping busy intervals.
    sort.Slice(busy, func(i, j int) bool { return busy[i].Start.Before(busy[j].Start) })
    merged := make([]Interval, 0, len(busy))
    for _, iv := range busy {
        if n := len(merged); n > 0 && !iv.Start.After(merged[n-1].End) {
            if iv.End.After(merged[n-1].End) {
                merged[n-1].End = iv.End
            }
            continue
        }
        merged = append(merged, iv)
    }

    // Find gaps of >= duration in [from, to].
    slots := make([]Interval, 0)
    cursor := from
    for _, iv := range merged {
        if iv.Start.After(cursor) && iv.Start.Sub(cursor) >= duration {
            slots = append(slots, Interval{Start: cursor, End: iv.Start})
        }
        if iv.End.After(cursor) {
            cursor = iv.End
        }
    }
    if to.Sub(cursor) >= duration {
        slots = append(slots, Interval{Start: cursor, End: to})
    }
    return slots, nil
}

using Npgsql;

namespace Calendar.FreeBusy;

public sealed class Interval(DateTimeOffset start, DateTimeOffset end)
{
    public DateTimeOffset Start { get; } = start;
    public DateTimeOffset End { get; set; } = end;   // mutable so merging can extend it
}

public sealed class FreeBusyService(NpgsqlDataSource dataSource)
{
    // Returns available slots of the given duration that are free for all users.
    public async Task<IReadOnlyList<Interval>> FindSlotsAsync(
        IReadOnlyList<Guid> users,
        DateTimeOffset from,
        DateTimeOffset to,
        TimeSpan duration,
        CancellationToken ct)
    {
        await using var conn = await dataSource.OpenConnectionAsync(ct);

        var busy = new List<Interval>();
        foreach (var u in users)
        {
            await using var cmd = new NpgsqlCommand(
                """
                SELECT starts_at, ends_at FROM free_busy_intervals
                WHERE user_id = @u AND ends_at > @from AND starts_at < @to
                """, conn);
            cmd.Parameters.AddWithValue("u", u);
            cmd.Parameters.AddWithValue("from", from);
            cmd.Parameters.AddWithValue("to", to);

            await using var reader = await cmd.ExecuteReaderAsync(ct);
            while (await reader.ReadAsync(ct))
            {
                busy.Add(new Interval(reader.GetFieldValue<DateTimeOffset>(0),
                                      reader.GetFieldValue<DateTimeOffset>(1)));
            }
        }

        // Merge overlapping busy intervals.
        busy.Sort((a, b) => a.Start.CompareTo(b.Start));
        var merged = new List<Interval>(busy.Count);
        foreach (var iv in busy)
        {
            if (merged.Count > 0 && iv.Start <= merged[^1].End)
            {
                if (iv.End > merged[^1].End)
                {
                    merged[^1].End = iv.End;
                }
                continue;
            }
            merged.Add(iv);
        }

        // Find gaps of >= duration in [from, to].
        var slots = new List<Interval>();
        var cursor = from;
        foreach (var iv in merged)
        {
            if (iv.Start > cursor && iv.Start - cursor >= duration)
            {
                slots.Add(new Interval(cursor, iv.Start));
            }
            if (iv.End > cursor)
            {
                cursor = iv.End;
            }
        }
        if (to - cursor >= duration)
        {
            slots.Add(new Interval(cursor, to));
        }
        return slots;
    }
}

from dataclasses import dataclass
from datetime import datetime, timedelta
from uuid import UUID

from psycopg_pool import AsyncConnectionPool

QUERY = """
    SELECT starts_at, ends_at FROM free_busy_intervals
    WHERE user_id = %s AND ends_at > %s AND starts_at < %s
"""

@dataclass(slots=True)
class Interval:
    start: datetime
    end: datetime   # mutable so merging can extend it

async def find_slots(
    pool: AsyncConnectionPool,
    users: list[UUID],
    from_: datetime,
    to: datetime,
    duration: timedelta,
) -> list[Interval]:
    """Return available slots of given duration that are free for all users."""
    busy: list[Interval] = []
    async with pool.connection() as conn, conn.cursor() as cur:
        for user_id in users:
            await cur.execute(QUERY, (user_id, from_, to))
            busy.extend(Interval(start, end) for start, end in await cur.fetchall())

    # Merge overlapping busy intervals.
    busy.sort(key=lambda iv: iv.start)
    merged: list[Interval] = []
    for iv in busy:
        if merged and iv.start <= merged[-1].end:
            merged[-1].end = max(merged[-1].end, iv.end)
            continue
        merged.append(iv)

    # Find gaps of >= duration in [from_, to].
    slots: list[Interval] = []
    cursor = from_
    for iv in merged:
        if iv.start > cursor and iv.start - cursor >= duration:
            slots.append(Interval(cursor, iv.start))
        cursor = max(cursor, iv.end)
    if to - cursor >= duration:
        slots.append(Interval(cursor, to))
    return slots