feat(backend): presence stats count time online from sessions in the browser time zone, capture uptime marks hours without data, presence reaches every stream of the account, message volume keeps empty days
This commit is contained in:
hh committed
1 parent
bb38e23f29
commit
f44597b2dc
14 files changed
+585
-26
No files matched your search
@@ -4,23 +4,31 @@ import asyncpg
|
||||
|
||||
from utils.read.accounts import self_user_id
|
||||
from utils.read.models import ResponseStats, VolumeBucket
|
||||
from utils.read.period import period_bounds
|
||||
|
||||
_VOLUME = """
|
||||
WITH counts AS (
|
||||
SELECT (date AT TIME ZONE $5)::date AS day, count(*) AS total,
|
||||
count(*) FILTER (WHERE sender_id = $3) AS outgoing
|
||||
FROM messages
|
||||
WHERE account_id = $1 AND chat_id = $2 AND date >= $4
|
||||
GROUP BY 1
|
||||
)
|
||||
SELECT local_day::date AS bucket, coalesce(total, 0) AS total,
|
||||
coalesce(outgoing, 0) AS outgoing
|
||||
FROM generate_series(($4 AT TIME ZONE $5)::date, (now() AT TIME ZONE $5)::date,
|
||||
interval '1 day') AS local_day
|
||||
LEFT JOIN counts ON counts.day = local_day::date
|
||||
ORDER BY 1
|
||||
"""
|
||||
|
||||
|
||||
async def message_volume(
|
||||
pool: asyncpg.Pool, account_id: int, chat_id: int, *, days: int = 90
|
||||
pool: asyncpg.Pool, account_id: int, chat_id: int, *, days: int, tz: str
|
||||
) -> list[VolumeBucket]:
|
||||
self_id = await self_user_id(pool, account_id)
|
||||
rows = await pool.fetch(
|
||||
"SELECT date_trunc('day', date) AS bucket, count(*) AS total, "
|
||||
"count(*) FILTER (WHERE sender_id = $3) AS outgoing "
|
||||
"FROM messages "
|
||||
"WHERE account_id = $1 AND chat_id = $2 AND date >= $4 "
|
||||
"GROUP BY bucket ORDER BY bucket",
|
||||
account_id,
|
||||
chat_id,
|
||||
self_id,
|
||||
datetime.now(UTC) - timedelta(days=days),
|
||||
)
|
||||
since, _ = await period_bounds(pool, days, tz)
|
||||
rows = await pool.fetch(_VOLUME, account_id, chat_id, self_id, since, tz)
|
||||
return [
|
||||
VolumeBucket(
|
||||
bucket=row["bucket"],
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from datetime import datetime
|
||||
from datetime import date, datetime
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel
|
||||
@@ -362,8 +362,61 @@ class PresenceHourly(BaseModel):
|
||||
last_seen: datetime | None
|
||||
|
||||
|
||||
class PresenceSession(BaseModel):
|
||||
started: datetime
|
||||
ended: datetime
|
||||
seconds: float
|
||||
online_seconds: float
|
||||
ongoing: bool
|
||||
|
||||
|
||||
class PresenceHour(BaseModel):
|
||||
weekday: int
|
||||
hour: int
|
||||
hours: int
|
||||
online_minutes: float | None
|
||||
covered_hours: float
|
||||
|
||||
|
||||
class PresenceDay(BaseModel):
|
||||
day: date
|
||||
online_seconds: float
|
||||
covered_seconds: float
|
||||
|
||||
|
||||
class CoverageGap(BaseModel):
|
||||
started: datetime
|
||||
ended: datetime
|
||||
|
||||
|
||||
class PresenceSummary(BaseModel):
|
||||
online_seconds: float
|
||||
daily_seconds: float | None
|
||||
covered_seconds: float
|
||||
session_count: int
|
||||
median_session_seconds: float | None
|
||||
first_online: int | None
|
||||
last_offline: int | None
|
||||
last_seen: datetime | None
|
||||
|
||||
|
||||
class PresenceStats(BaseModel):
|
||||
peer_id: int
|
||||
tz: str
|
||||
since: datetime
|
||||
until: datetime
|
||||
current: PresenceSample | None
|
||||
online: bool
|
||||
hidden: bool
|
||||
summary: PresenceSummary
|
||||
week: list[PresenceHour]
|
||||
days: list[PresenceDay]
|
||||
gaps: list[CoverageGap]
|
||||
sessions: list[PresenceSession]
|
||||
|
||||
|
||||
class VolumeBucket(BaseModel):
|
||||
bucket: datetime
|
||||
bucket: date
|
||||
total: int
|
||||
outgoing: int
|
||||
incoming: int
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
from datetime import datetime
|
||||
|
||||
import asyncpg
|
||||
|
||||
DEFAULT_TZ = "UTC"
|
||||
MAX_DAYS = 365
|
||||
|
||||
_timezones: set[str] = set()
|
||||
|
||||
|
||||
async def known_timezone(pool: asyncpg.Pool, tz: str) -> bool:
|
||||
if not _timezones:
|
||||
rows = await pool.fetch("SELECT name FROM pg_timezone_names")
|
||||
_timezones.update(row["name"] for row in rows)
|
||||
return tz in _timezones
|
||||
|
||||
|
||||
async def period_bounds(
|
||||
pool: asyncpg.Pool, days: int, tz: str
|
||||
) -> tuple[datetime, datetime]:
|
||||
row = await pool.fetchrow(
|
||||
"SELECT (date_trunc('day', now() AT TIME ZONE $1) "
|
||||
"- make_interval(days => $2 - 1)) AT TIME ZONE $1 AS since, now() AS until",
|
||||
tz,
|
||||
days,
|
||||
)
|
||||
return row["since"], row["until"]
|
||||
@@ -0,0 +1,289 @@
|
||||
import json
|
||||
|
||||
import asyncpg
|
||||
|
||||
from utils.read.models import (
|
||||
CoverageGap,
|
||||
Page,
|
||||
PresenceDay,
|
||||
PresenceHour,
|
||||
PresenceSession,
|
||||
PresenceStats,
|
||||
PresenceSummary,
|
||||
)
|
||||
from utils.read.period import period_bounds
|
||||
from utils.read.presence import current_presence
|
||||
|
||||
SECONDS_IN_HOUR = 3600
|
||||
SECONDS_IN_DAY = 86_400
|
||||
MIN_CELL_COVERAGE = 1800
|
||||
MIN_CELL_SHARE = 0.5
|
||||
STATS_WORK_MEM = "32MB"
|
||||
FUZZY_STATUSES = {"recently", "last_week", "last_month", "long_time_ago"}
|
||||
|
||||
_SESSIONS = """
|
||||
marks AS (
|
||||
SELECT ts, status,
|
||||
least(next_offline_date, ts + interval '5 minutes',
|
||||
lead(coalesce(last_online_date, ts)) OVER w, $4) AS ended,
|
||||
lag(CASE WHEN status = 'online'
|
||||
THEN least(next_offline_date, ts + interval '5 minutes')
|
||||
ELSE coalesce(last_online_date, ts) END) OVER w AS previous
|
||||
FROM presence
|
||||
WHERE account_id = $1 AND peer_id = $2
|
||||
AND ts >= $3::timestamptz - interval '5 minutes' AND ts <= $4
|
||||
WINDOW w AS (ORDER BY ts)
|
||||
),
|
||||
online AS (
|
||||
SELECT greatest(ts, $3) AS started, ended,
|
||||
count(*) FILTER (WHERE previous IS NULL
|
||||
OR ts > previous + interval '5 minutes')
|
||||
OVER (ORDER BY ts) AS visit
|
||||
FROM marks
|
||||
WHERE status = 'online' AND ended > greatest(ts, $3)
|
||||
),
|
||||
sessions AS (
|
||||
SELECT min(started) AS started, max(ended) AS ended,
|
||||
extract(epoch FROM sum(ended - started))::float8 AS online_seconds
|
||||
FROM online
|
||||
GROUP BY visit
|
||||
)
|
||||
"""
|
||||
|
||||
_SESSION_FIELDS = """
|
||||
started, ended, extract(epoch FROM ended - started)::float8 AS seconds,
|
||||
online_seconds, ended >= $4 AS ongoing
|
||||
"""
|
||||
|
||||
_SESSION_PAGE = f"""
|
||||
WITH {_SESSIONS}
|
||||
SELECT {_SESSION_FIELDS}
|
||||
FROM sessions
|
||||
ORDER BY started DESC
|
||||
LIMIT $5 OFFSET $6
|
||||
""" # noqa: S608
|
||||
|
||||
_COVERED_HOURS = """
|
||||
SELECT local_hour,
|
||||
extract(epoch FROM sum(
|
||||
least(ended, (local_hour + interval '1 hour') AT TIME ZONE $5)
|
||||
- greatest(started, local_hour AT TIME ZONE $5)))::float8 AS seconds
|
||||
FROM uptime,
|
||||
generate_series(date_trunc('hour', started AT TIME ZONE $5),
|
||||
date_trunc('hour', ended AT TIME ZONE $5),
|
||||
interval '1 hour') AS local_hour
|
||||
GROUP BY local_hour
|
||||
"""
|
||||
|
||||
_STATS = f"""
|
||||
WITH {_SESSIONS},
|
||||
uptime AS (
|
||||
SELECT greatest(started_at, $3) AS started,
|
||||
least(seen_at + interval '90 seconds', $4) AS ended
|
||||
FROM capture_uptime
|
||||
WHERE account_id = $1 AND started_at < $4
|
||||
AND seen_at + interval '90 seconds' > $3
|
||||
),
|
||||
quarters AS (
|
||||
SELECT quarter, sum(ended - started - spill) AS inside, sum(spill) AS spill
|
||||
FROM (SELECT started, ended, quarter,
|
||||
greatest(ended - quarter - interval '15 minutes',
|
||||
interval '0') AS spill
|
||||
FROM (SELECT started, ended,
|
||||
date_bin('15 minutes', started, timestamptz 'epoch') AS quarter
|
||||
FROM online) binned) split
|
||||
GROUP BY quarter
|
||||
),
|
||||
online_hours AS (
|
||||
SELECT date_trunc('hour', quarter AT TIME ZONE $5) AS local_hour,
|
||||
extract(epoch FROM sum(share))::float8 AS seconds
|
||||
FROM (SELECT quarter, inside AS share FROM quarters
|
||||
UNION ALL
|
||||
SELECT quarter + interval '15 minutes', spill FROM quarters) shares
|
||||
GROUP BY 1
|
||||
),
|
||||
covered_hours AS ({_COVERED_HOURS}),
|
||||
hours AS (
|
||||
SELECT local_hour,
|
||||
extract(epoch FROM
|
||||
least((local_hour + interval '1 hour') AT TIME ZONE $5, $4)
|
||||
- greatest(local_hour AT TIME ZONE $5, $3))::float8 AS span,
|
||||
coalesce(online.seconds, 0) AS online,
|
||||
least(3600, greatest(coalesce(covered.seconds, 0),
|
||||
coalesce(online.seconds, 0))) AS covered
|
||||
FROM generate_series(date_trunc('hour', $3 AT TIME ZONE $5),
|
||||
date_trunc('hour', $4 AT TIME ZONE $5),
|
||||
interval '1 hour') AS local_hour
|
||||
LEFT JOIN online_hours AS online USING (local_hour)
|
||||
LEFT JOIN covered_hours AS covered USING (local_hour)
|
||||
),
|
||||
week AS (
|
||||
SELECT extract(isodow FROM local_hour)::int - 1 AS weekday,
|
||||
extract(hour FROM local_hour)::int AS hour,
|
||||
count(*) AS hours, sum(span) AS span,
|
||||
sum(online) AS online, sum(covered) AS covered
|
||||
FROM hours
|
||||
GROUP BY 1, 2
|
||||
),
|
||||
days AS (
|
||||
SELECT local_hour::date AS day, sum(online) AS online, sum(covered) AS covered
|
||||
FROM hours
|
||||
GROUP BY 1
|
||||
),
|
||||
uptime_reach AS (
|
||||
SELECT started,
|
||||
coalesce(max(ended) OVER (ORDER BY started
|
||||
ROWS BETWEEN UNBOUNDED PRECEDING
|
||||
AND 1 PRECEDING), $3) AS reached
|
||||
FROM uptime
|
||||
),
|
||||
gaps AS (
|
||||
SELECT reached AS started, started AS ended FROM uptime_reach
|
||||
UNION ALL
|
||||
SELECT coalesce(max(ended), $3), $4 FROM uptime
|
||||
),
|
||||
quiet AS (
|
||||
SELECT hour
|
||||
FROM week
|
||||
GROUP BY hour
|
||||
ORDER BY sum(online) / nullif(sum(covered), 0) NULLS LAST, hour
|
||||
LIMIT 1
|
||||
),
|
||||
activity_days AS (
|
||||
SELECT (sessions.started AT TIME ZONE $5) - make_interval(hours => quiet.hour)
|
||||
AS started,
|
||||
(sessions.ended AT TIME ZONE $5) - make_interval(hours => quiet.hour)
|
||||
AS ended,
|
||||
(($4 AT TIME ZONE $5) - make_interval(hours => quiet.hour))::date AS today
|
||||
FROM sessions, quiet
|
||||
),
|
||||
daily AS (
|
||||
SELECT extract(epoch FROM min(started - started::date))::float8 AS first_online,
|
||||
extract(epoch FROM max(ended - started::date))::float8 AS last_offline
|
||||
FROM activity_days
|
||||
WHERE started::date < today
|
||||
GROUP BY started::date
|
||||
)
|
||||
SELECT
|
||||
(SELECT count(*) FROM sessions) AS session_count,
|
||||
(SELECT coalesce(sum(online_seconds), 0) FROM sessions) AS online_seconds,
|
||||
(SELECT sum(covered) FROM hours) AS covered_seconds,
|
||||
(SELECT percentile_cont(0.5) WITHIN GROUP (
|
||||
ORDER BY extract(epoch FROM ended - started)) FROM sessions) AS median_session,
|
||||
(SELECT max(ended) FROM sessions) AS last_seen,
|
||||
(SELECT hour FROM quiet) AS day_start,
|
||||
(SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY first_online)
|
||||
FROM daily) AS first_online,
|
||||
(SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY last_offline)
|
||||
FROM daily) AS last_offline,
|
||||
(SELECT json_agg(json_build_object(
|
||||
'weekday', weekday, 'hour', hour, 'hours', hours, 'span', span,
|
||||
'online', online, 'covered', covered))
|
||||
FROM week) AS week,
|
||||
(SELECT json_agg(json_build_object(
|
||||
'day', day, 'online_seconds', online, 'covered_seconds', covered)
|
||||
ORDER BY day)
|
||||
FROM days) AS days,
|
||||
(SELECT json_agg(json_build_object('started', started, 'ended', ended)
|
||||
ORDER BY started)
|
||||
FROM gaps WHERE ended - started > interval '5 minutes') AS gaps,
|
||||
(SELECT json_agg(row_to_json(page) ORDER BY page.started DESC)
|
||||
FROM (SELECT {_SESSION_FIELDS} FROM sessions
|
||||
ORDER BY started DESC LIMIT $6 OFFSET $7) page) AS sessions
|
||||
""" # noqa: S608
|
||||
|
||||
|
||||
def _time_of_day(shifted: float | None, day_start: int | None) -> int | None:
|
||||
if shifted is None or day_start is None:
|
||||
return None
|
||||
return round(shifted + day_start * SECONDS_IN_HOUR) % SECONDS_IN_DAY
|
||||
|
||||
|
||||
def _week(rows: list[dict]) -> list[PresenceHour]:
|
||||
cells = {(row["weekday"], row["hour"]): row for row in rows}
|
||||
week: list[PresenceHour] = []
|
||||
for weekday in range(7):
|
||||
for hour in range(24):
|
||||
row = cells.get(
|
||||
(weekday, hour), {"hours": 0, "span": 0, "online": 0, "covered": 0}
|
||||
)
|
||||
covered = row["covered"]
|
||||
minutes = (
|
||||
min(60.0, row["online"] / covered * 60)
|
||||
if covered > 0
|
||||
and covered >= min(MIN_CELL_COVERAGE, row["span"] * MIN_CELL_SHARE)
|
||||
else None
|
||||
)
|
||||
week.append(
|
||||
PresenceHour(
|
||||
weekday=weekday,
|
||||
hour=hour,
|
||||
hours=row["hours"],
|
||||
online_minutes=minutes,
|
||||
covered_hours=covered / SECONDS_IN_HOUR,
|
||||
)
|
||||
)
|
||||
return week
|
||||
|
||||
|
||||
async def presence_stats( # noqa: PLR0913
|
||||
pool: asyncpg.Pool, account_id: int, peer_id: int, page: Page, *, days: int, tz: str
|
||||
) -> PresenceStats:
|
||||
since, until = await period_bounds(pool, days, tz)
|
||||
async with pool.acquire() as conn, conn.transaction():
|
||||
await conn.execute(f"SET LOCAL work_mem = '{STATS_WORK_MEM}'")
|
||||
row = await conn.fetchrow(
|
||||
_STATS,
|
||||
account_id,
|
||||
peer_id,
|
||||
since,
|
||||
until,
|
||||
tz,
|
||||
page.capped_limit,
|
||||
page.offset,
|
||||
)
|
||||
current = await current_presence(pool, account_id, peer_id)
|
||||
sessions = [PresenceSession(**item) for item in json.loads(row["sessions"] or "[]")]
|
||||
covered = row["covered_seconds"] or 0
|
||||
online = row["online_seconds"]
|
||||
return PresenceStats(
|
||||
peer_id=peer_id,
|
||||
tz=tz,
|
||||
since=since,
|
||||
until=until,
|
||||
current=current,
|
||||
online=current is not None
|
||||
and current.status == "online"
|
||||
and current.next_offline_date is not None
|
||||
and current.next_offline_date > until,
|
||||
hidden=row["session_count"] == 0
|
||||
and current is not None
|
||||
and current.status in FUZZY_STATUSES,
|
||||
summary=PresenceSummary(
|
||||
online_seconds=online,
|
||||
daily_seconds=online / covered * SECONDS_IN_DAY
|
||||
if covered >= SECONDS_IN_HOUR
|
||||
else None,
|
||||
covered_seconds=covered,
|
||||
session_count=row["session_count"],
|
||||
median_session_seconds=row["median_session"],
|
||||
first_online=_time_of_day(row["first_online"], row["day_start"]),
|
||||
last_offline=_time_of_day(row["last_offline"], row["day_start"]),
|
||||
last_seen=row["last_seen"]
|
||||
or (current.last_online_date if current else None),
|
||||
),
|
||||
week=_week(json.loads(row["week"] or "[]")),
|
||||
days=[PresenceDay(**item) for item in json.loads(row["days"] or "[]")],
|
||||
gaps=[CoverageGap(**item) for item in json.loads(row["gaps"] or "[]")],
|
||||
sessions=sessions,
|
||||
)
|
||||
|
||||
|
||||
async def presence_sessions( # noqa: PLR0913
|
||||
pool: asyncpg.Pool, account_id: int, peer_id: int, page: Page, *, days: int, tz: str
|
||||
) -> list[PresenceSession]:
|
||||
since, until = await period_bounds(pool, days, tz)
|
||||
rows = await pool.fetch(
|
||||
_SESSION_PAGE, account_id, peer_id, since, until, page.capped_limit, page.offset
|
||||
)
|
||||
return [PresenceSession(**dict(row)) for row in rows]
|
||||
Reference in new issue
Block a user