✨ Add stats on conversations (#828)

* ✨ Add stats on conversations

* 🩹 Fix conversation stats compute

* ⚡️ Slightly better index

* 🔊

* 🩹 Some fixes

* 🐛 Fix aggregation query

* 🐛 Fix $merge stage of query

* 🐛 Fix query on sessionId

* ✨ Compute weekly/monthly unique users

* 🩹 Final tweaks

* ⚡️ Split aggregations for better perf

No need to recompute DAUs / WAUs and so on

* ♻️ Deprecate PARQUET_EXPORT_SECRET & fix sec vuln

* Add ADMIN_API_SECRET to CD action

---------

Co-authored-by: Nathan Sarrazin <sarrazin.nathan@gmail.com>

.env

@@ -109,7 +109,9 @@ PUBLIC_ANNOUNCEMENT_BANNERS=`[
 
 PARQUET_EXPORT_DATASET=
 PARQUET_EXPORT_HF_TOKEN=
-PARQUET_EXPORT_SECRET=
+ADMIN_API_SECRET=# secret to admin API calls, like computing usage stats or exporting parquet data
+
+PARQUET_EXPORT_SECRET=#DEPRECATED, use ADMIN_API_SECRET instead
 
 RATE_LIMIT= # requests per minute
 MESSAGES_BEFORE_LOGIN=# how many messages a user can send in a conversation before having to login. set to 0 to force login right away

.github/workflows/deploy-release.yml

@@ -26,6 +26,7 @@ jobs:
           MONGODB_URL: ${{ secrets.MONGODB_URL }}
           HF_DEPLOYMENT_TOKEN: ${{ secrets.HF_DEPLOYMENT_TOKEN }}
           WEBHOOK_URL_REPORT_ASSISTANT: ${{ secrets.WEBHOOK_URL_REPORT_ASSISTANT }}
+          ADMIN_API_SECRET: ${{ secrets.ADMIN_API_SECRET }}
         run: npm run updateProdEnv
   sync-to-hub:
     runs-on: ubuntu-latest

package-lock.json

@@ -4093,10 +4093,17 @@
 			"resolved": "https://registry.npmjs.org/int53/-/int53-0.2.4.tgz",
 			"integrity": "sha512-a5jlKftS7HUOhkUyYD7j2sJ/ZnvWiNlZS1ldR+g1ifQ+/UuZXIE+YTc/lK1qGj/GwAU5F8Z0e1eVq2t1J5Ob2g=="
 		},
-		"node_modules/ip": {
-			"version": "2.0.0",
-			"resolved": "https://registry.npmjs.org/ip/-/ip-2.0.0.tgz",
-			"integrity": "sha512-WKa+XuLG1A1R0UWhl2+1XQSi+fZWMsYKffMZTTYsiZaUD8k2yDAj5atimTUD2TZkyCkNEeYE5NhFZmupOGtjYQ=="
+		"node_modules/ip-address": {
+			"version": "9.0.5",
+			"resolved": "https://registry.npmjs.org/ip-address/-/ip-address-9.0.5.tgz",
+			"integrity": "sha512-zHtQzGojZXTwZTHQqra+ETKd4Sn3vgi7uBmlPoXVWZqYvuKmtI0l/VZTjqGmJY9x88GGOaZ9+G9ES8hC4T4X8g==",
+			"dependencies": {
+				"jsbn": "1.1.0",
+				"sprintf-js": "^1.1.3"
+			},
+			"engines": {
+				"node": ">= 12"
+			}
 		},
 		"node_modules/is-arrayish": {
 			"version": "0.3.2",
@@ -4267,6 +4274,11 @@
 				"js-yaml": "bin/js-yaml.js"
 			}
 		},
+		"node_modules/jsbn": {
+			"version": "1.1.0",
+			"resolved": "https://registry.npmjs.org/jsbn/-/jsbn-1.1.0.tgz",
+			"integrity": "sha512-4bYVV3aAMtDTTu4+xsDYa6sy9GyJ69/amsu9sYF2zqjiEoZA5xJi3BrfX3uY+/IekIu7MwdObdbDWpoZdBv3/A=="
+		},
 		"node_modules/jsdom": {
 			"version": "22.0.0",
 			"resolved": "https://registry.npmjs.org/jsdom/-/jsdom-22.0.0.tgz",
@@ -6294,15 +6306,15 @@
 			"integrity": "sha512-YIK6I2lsH072UE0aOFxxY1dPDCS43I5ktqHpeAsuLNYWkE5pGxRGWfDM4/vSUfNzXjC1Ivzt3qx31PCLmc9yqg=="
 		},
 		"node_modules/socks": {
-			"version": "2.7.1",
-			"resolved": "https://registry.npmjs.org/socks/-/socks-2.7.1.tgz",
-			"integrity": "sha512-7maUZy1N7uo6+WVEX6psASxtNlKaNVMlGQKkG/63nEDdLOWNbiUMoLK7X4uYoLhQstau72mLgfEWcXcwsaHbYQ==",
+			"version": "2.7.3",
+			"resolved": "https://registry.npmjs.org/socks/-/socks-2.7.3.tgz",
+			"integrity": "sha512-vfuYK48HXCTFD03G/1/zkIls3Ebr2YNa4qU9gHDZdblHLiqhJrJGkY3+0Nx0JpN9qBhJbVObc1CNciT1bIZJxw==",
 			"dependencies": {
-				"ip": "^2.0.0",
+				"ip-address": "^9.0.5",
 				"smart-buffer": "^4.2.0"
 			},
 			"engines": {
-				"node": ">= 10.13.0",
+				"node": ">= 10.0.0",
 				"npm": ">= 3.0.0"
 			}
 		},
@@ -6345,6 +6357,11 @@
 				"memory-pager": "^1.0.2"
 			}
 		},
+		"node_modules/sprintf-js": {
+			"version": "1.1.3",
+			"resolved": "https://registry.npmjs.org/sprintf-js/-/sprintf-js-1.1.3.tgz",
+			"integrity": "sha512-Oo+0REFV59/rz3gfJNKQiBlwfHaSESl1pcGyABQsnnIfWOFt6JNj5gCog2U6MLZ//IGYD+nA8nI+mTShREReaA=="
+		},
 		"node_modules/stackback": {
 			"version": "0.0.2",
 			"resolved": "https://registry.npmjs.org/stackback/-/stackback-0.0.2.tgz",

scripts/updateProdEnv.ts

@@ -7,6 +7,7 @@ const OPENID_CONFIG = process.env.OPENID_CONFIG;
 const MONGODB_URL = process.env.MONGODB_URL;
 const HF_TOKEN = process.env.HF_TOKEN ?? process.env.HF_ACCESS_TOKEN; // token used for API requests in prod
 const WEBHOOK_URL_REPORT_ASSISTANT = process.env.WEBHOOK_URL_REPORT_ASSISTANT; // slack webhook url used to get "report assistant" events
+const ADMIN_API_SECRET = process.env.ADMIN_API_SECRET;
 
 // Read the content of the file .env.template
 const PUBLIC_CONFIG = fs.readFileSync(".env.template", "utf8");
@@ -18,6 +19,7 @@ OPENID_CONFIG=${OPENID_CONFIG}
 SERPER_API_KEY=${SERPER_API_KEY}
 HF_TOKEN=${HF_TOKEN}
 WEBHOOK_URL_REPORT_ASSISTANT=${WEBHOOK_URL_REPORT_ASSISTANT}
+ADMIN_API_SECRET=${ADMIN_API_SECRET}
 `;
 
 // Make an HTTP POST request to add the space secrets

src/hooks.server.ts

@@ -1,4 +1,10 @@
-import { COOKIE_NAME, EXPOSE_API, MESSAGES_BEFORE_LOGIN } from "$env/static/private";
+import {
+	ADMIN_API_SECRET,
+	COOKIE_NAME,
+	EXPOSE_API,
+	MESSAGES_BEFORE_LOGIN,
+	PARQUET_EXPORT_SECRET,
+} from "$env/static/private";
 import type { Handle } from "@sveltejs/kit";
 import {
 	PUBLIC_GOOGLE_ANALYTICS_ID,
@@ -29,6 +35,18 @@ export const handle: Handle = async ({ event, resolve }) => {
 		});
 	}
 
+	if (event.url.pathname.startsWith(`${base}/admin/`) || event.url.pathname === `${base}/admin`) {
+		const ADMIN_SECRET = ADMIN_API_SECRET || PARQUET_EXPORT_SECRET;
+
+		if (!ADMIN_SECRET) {
+			return errorResponse(500, "Admin API is not configured");
+		}
+
+		if (event.request.headers.get("Authorization") !== `Bearer ${ADMIN_SECRET}`) {
+			return errorResponse(401, "Unauthorized");
+		}
+	}
+
 	const token = event.cookies.get(COOKIE_NAME);
 
 	let secretSessionId: string;

src/lib/server/database.ts

@@ -9,6 +9,7 @@ import type { MessageEvent } from "$lib/types/MessageEvent";
 import type { Session } from "$lib/types/Session";
 import type { Assistant } from "$lib/types/Assistant";
 import type { Report } from "$lib/types/Report";
+import type { ConversationStats } from "$lib/types/ConversationStats";
 
 if (!MONGODB_URL) {
 	throw new Error(
@@ -24,7 +25,10 @@ export const connectPromise = client.connect().catch(console.error);
 
 const db = client.db(MONGODB_DB_NAME + (import.meta.env.MODE === "test" ? "-test" : ""));
 
+export const CONVERSATION_STATS_COLLECTION = "conversations.stats";
+
 const conversations = db.collection<Conversation>("conversations");
+const conversationStats = db.collection<ConversationStats>(CONVERSATION_STATS_COLLECTION);
 const assistants = db.collection<Assistant>("assistants");
 const reports = db.collection<Report>("reports");
 const sharedConversations = db.collection<SharedConversation>("sharedConversations");
@@ -38,6 +42,7 @@ const bucket = new GridFSBucket(db, { bucketName: "files" });
 export { client, db };
 export const collections = {
 	conversations,
+	conversationStats,
 	assistants,
 	reports,
 	sharedConversations,
@@ -68,6 +73,33 @@ client.on("open", () => {
 			{ partialFilterExpression: { userId: { $exists: true } } }
 		)
 		.catch(console.error);
+	// To do stats on conversations
+	conversations.createIndex({ updatedAt: 1 }).catch(console.error);
+	// Not strictly necessary, could use _id, but more convenient. Also for stats
+	conversations.createIndex({ createdAt: 1 }).catch(console.error);
+	// To do stats on conversation messages
+	conversations.createIndex({ "messages.createdAt": 1 }, { sparse: true }).catch(console.error);
+	// Unique index for stats
+	conversationStats
+		.createIndex(
+			{
+				type: 1,
+				"date.field": 1,
+				"date.span": 1,
+				"date.at": 1,
+				distinct: 1,
+			},
+			{ unique: true }
+		)
+		.catch(console.error);
+	// Allow easy check of last computed stat for given type/dateField
+	conversationStats
+		.createIndex({
+			type: 1,
+			"date.field": 1,
+			"date.at": 1,
+		})
+		.catch(console.error);
 	abortedGenerations.createIndex({ updatedAt: 1 }, { expireAfterSeconds: 30 }).catch(console.error);
 	abortedGenerations.createIndex({ conversationId: 1 }, { unique: true }).catch(console.error);
 	sharedConversations.createIndex({ hash: 1 }, { unique: true }).catch(console.error);

src/lib/types/ConversationStats.ts

@@ -0,0 +1,13 @@
+import type { Timestamps } from "./Timestamps";
+
+export interface ConversationStats extends Timestamps {
+	date: {
+		at: Date;
+		span: "day" | "week" | "month";
+		field: "updatedAt" | "createdAt";
+	};
+	type: "conversation" | "message";
+	/**  _id => number of conversations/messages in the month */
+	distinct: "sessionId" | "userId" | "userOrSessionId" | "_id";
+	count: number;
+}

src/routes/admin/export/+server.ts

@@ -1,8 +1,4 @@
-import {
-	PARQUET_EXPORT_DATASET,
-	PARQUET_EXPORT_HF_TOKEN,
-	PARQUET_EXPORT_SECRET,
-} from "$env/static/private";
+import { PARQUET_EXPORT_DATASET, PARQUET_EXPORT_HF_TOKEN } from "$env/static/private";
 import { collections } from "$lib/server/database";
 import type { Message } from "$lib/types/Message";
 import { error } from "@sveltejs/kit";
@@ -13,17 +9,13 @@ import parquet from "parquetjs";
 import { z } from "zod";
 
 // Triger like this:
-// curl -X POST "http://localhost:5173/chat/admin/export" -H "Authorization: Bearer <PARQUET_EXPORT_SECRET>" -H "Content-Type: application/json" -d '{"model": "OpenAssistant/oasst-sft-6-llama-30b-xor"}'
+// curl -X POST "http://localhost:5173/chat/admin/export" -H "Authorization: Bearer <ADMIN_API_SECRET>" -H "Content-Type: application/json" -d '{"model": "OpenAssistant/oasst-sft-6-llama-30b-xor"}'
 
 export async function POST({ request }) {
-	if (!PARQUET_EXPORT_SECRET || !PARQUET_EXPORT_DATASET || !PARQUET_EXPORT_HF_TOKEN) {
+	if (!PARQUET_EXPORT_DATASET || !PARQUET_EXPORT_HF_TOKEN) {
 		throw error(500, "Parquet export is not configured.");
 	}
 
-	if (request.headers.get("Authorization") !== `Bearer ${PARQUET_EXPORT_SECRET}`) {
-		throw error(403);
-	}
-
 	const { model } = z
 		.object({
 			model: z.string(),

src/routes/admin/stats/compute/+server.ts

@@ -0,0 +1,217 @@
+import { json } from "@sveltejs/kit";
+import type { ConversationStats } from "$lib/types/ConversationStats";
+import { CONVERSATION_STATS_COLLECTION, collections } from "$lib/server/database.js";
+
+// Triger like this:
+// curl -X POST "http://localhost:5173/chat/admin/stats/compute" -H "Authorization: Bearer <ADMIN_API_SECRET>"
+
+export async function POST() {
+	for (const span of ["day", "week", "month"] as const) {
+		computeStats({ dateField: "updatedAt", type: "conversation", span }).catch(console.error);
+		computeStats({ dateField: "createdAt", type: "conversation", span }).catch(console.error);
+		computeStats({ dateField: "createdAt", type: "message", span }).catch(console.error);
+	}
+
+	return json({}, { status: 202 });
+}
+
+async function computeStats(params: {
+	dateField: ConversationStats["date"]["field"];
+	span: ConversationStats["date"]["span"];
+	type: ConversationStats["type"];
+}) {
+	const lastComputed = await collections.conversationStats.findOne(
+		{ "date.field": params.dateField, "date.span": params.span, type: params.type },
+		{ sort: { "date.at": -1 } }
+	);
+
+	// If the last computed week is at the beginning of the last computed month, we need to include some days from the previous month
+	// In those cases we need to compute the stats from before the last month as everything is one aggregation
+	const minDate = lastComputed ? lastComputed.date.at : new Date(0);
+
+	console.log("Computing stats for", params.type, params.span, params.dateField, "from", minDate);
+
+	const dateField = params.type === "message" ? "messages." + params.dateField : params.dateField;
+
+	const pipeline = [
+		{
+			$match: {
+				[dateField]: { $gte: minDate },
+			},
+		},
+		{
+			$project: {
+				[dateField]: 1,
+				sessionId: 1,
+				userId: 1,
+			},
+		},
+		...(params.type === "message"
+			? [
+					{
+						$unwind: "$messages",
+					},
+					{
+						$match: {
+							[dateField]: { $gte: minDate },
+						},
+					},
+			  ]
+			: []),
+		{
+			$sort: {
+				[dateField]: 1,
+			},
+		},
+		{
+			$facet: {
+				userId: [
+					{
+						$match: {
+							userId: { $exists: true },
+						},
+					},
+					{
+						$group: {
+							_id: {
+								at: { $dateTrunc: { date: `$${dateField}`, unit: params.span } },
+								userId: "$userId",
+							},
+						},
+					},
+					{
+						$group: {
+							_id: "$_id.at",
+							count: { $sum: 1 },
+						},
+					},
+					{
+						$project: {
+							_id: 0,
+							date: {
+								at: "$_id",
+								field: params.dateField,
+								span: params.span,
+							},
+							distinct: "userId",
+							count: 1,
+						},
+					},
+				],
+				sessionId: [
+					{
+						$match: {
+							sessionId: { $exists: true },
+						},
+					},
+					{
+						$group: {
+							_id: {
+								at: { $dateTrunc: { date: `$${dateField}`, unit: params.span } },
+								sessionId: "$sessionId",
+							},
+						},
+					},
+					{
+						$group: {
+							_id: "$_id.at",
+							count: { $sum: 1 },
+						},
+					},
+					{
+						$project: {
+							_id: 0,
+							date: {
+								at: "$_id",
+								field: params.dateField,
+								span: params.span,
+							},
+							distinct: "sessionId",
+							count: 1,
+						},
+					},
+				],
+				userOrSessionId: [
+					{
+						$group: {
+							_id: {
+								at: { $dateTrunc: { date: `$${dateField}`, unit: params.span } },
+								userOrSessionId: { $ifNull: ["$userId", "$sessionId"] },
+							},
+						},
+					},
+					{
+						$group: {
+							_id: "$_id.at",
+							count: { $sum: 1 },
+						},
+					},
+					{
+						$project: {
+							_id: 0,
+							date: {
+								at: "$_id",
+								field: params.dateField,
+								span: params.span,
+							},
+							distinct: "userOrSessionId",
+							count: 1,
+						},
+					},
+				],
+				_id: [
+					{
+						$group: {
+							_id: { $dateTrunc: { date: `$${dateField}`, unit: params.span } },
+							count: { $sum: 1 },
+						},
+					},
+					{
+						$project: {
+							_id: 0,
+							date: {
+								at: "$_id",
+								field: params.dateField,
+								span: params.span,
+							},
+							distinct: "_id",
+							count: 1,
+						},
+					},
+				],
+			},
+		},
+		{
+			$project: {
+				stats: {
+					$concatArrays: ["$userId", "$sessionId", "$userOrSessionId", "$_id"],
+				},
+			},
+		},
+		{
+			$unwind: "$stats",
+		},
+		{
+			$replaceRoot: {
+				newRoot: "$stats",
+			},
+		},
+		{
+			$set: {
+				type: params.type,
+			},
+		},
+		{
+			$merge: {
+				into: CONVERSATION_STATS_COLLECTION,
+				on: ["date.at", "type", "date.span", "date.field", "distinct"],
+				whenMatched: "replace",
+				whenNotMatched: "insert",
+			},
+		},
+	];
+
+	await collections.conversations.aggregate(pipeline, { allowDiskUse: true }).next();
+
+	console.log("Computed stats for", params.type, params.span, params.dateField);
+}