Skip to content

Batch processing for Longest Listening Session #555

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
webysther wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Batch processing for Longest Listening Session #555

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b92853e
Batch process the getLongestListeningSession
webysther Sep 12, 2025
2b37491
TYPO
webysther Sep 12, 2025
c047777
Merge branch 'Yooooomi:master' into master
webysther Jan 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
214 changes: 122 additions & 92 deletions apps/server/src/database/queries/stats.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -671,116 +671,146 @@ export const getLongestListeningSession = async (
end: Date,
) => {
const sessionBreakThreshold = 10 * 60 * 1000;
const subtract = {
$cond: [
"$$value.last",
{
$subtract: [
"$$this.played_at",
{
$add: ["$$value.last.played_at", "$$value.last.durationMs"],
},
],
},
sessionBreakThreshold + 1,
],
};
const yearsStep = 1;

Copilot AI Jan 3, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The variable name 'yearsStep' is defined but could be more descriptive. Consider renaming it to 'batchSizeYears' or 'yearsBatchSize' to better convey that it represents the batch size in years for processing the data.

Copilot uses AI. Check for mistakes.

const item = { subtract, info: "$$this" };
const startDate = new Date(start);
const endDate = new Date(end);

const longestSessions = await InfosModel.aggregate([
...basicMatch(userId, start, end),
{ $sort: { played_at: 1 } },
{
$group: {
_id: "$owner",
infos: { $push: "$$ROOT" },
},
},
{
$addFields: {
distanceToLast: {
$reduce: {
input: "$infos",
initialValue: { distance: [], current: [] },
in: {
distance: {
$concatArrays: [
"$$value.distance",
{
$cond: {
if: {
$gt: [subtract, sessionBreakThreshold],
if (isNaN(startDate.getTime()) || isNaN(endDate.getTime()) || startDate >= endDate) {
return [];
}

const windows: { from: Date; to: Date }[] = [];
let cursor = new Date(startDate);
while (cursor < endDate) {
const next = new Date(cursor);
next.setFullYear(next.getFullYear() + yearsStep);
if (next > endDate) next.setTime(endDate.getTime());
windows.push({ from: new Date(cursor), to: new Date(next) });

Copilot AI Jan 3, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sessions that span across year boundaries may be incorrectly split into separate sessions. When a listening session starts in one year batch and continues into the next, the current implementation will treat them as two separate sessions instead of one continuous session. Consider implementing logic to handle cross-boundary sessions by either checking the last session of a batch against the first session of the next batch, or by overlapping the batch windows by the sessionBreakThreshold duration.

Suggested change
windows.push({ from: new Date(cursor), to: new Date(next) });
// Overlap windows by sessionBreakThreshold to avoid splitting sessions
const isFirstWindow = windows.length === 0;
const overlapFrom = isFirstWindow
? new Date(cursor)
: new Date(Math.max(startDate.getTime(), cursor.getTime() - sessionBreakThreshold));
windows.push({ from: overlapFrom, to: new Date(next) });

Copilot uses AI. Check for mistakes.
cursor = next;
}
Comment on lines +674 to +691

Copilot AI Jan 3, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The batching logic lacks documentation explaining why data is processed in 1-year chunks. Consider adding a comment that explains this design decision, particularly noting that it's a performance optimization to handle large datasets (as mentioned in the PR description, it fixes timeout issues for datasets larger than 7 years).

Copilot uses AI. Check for mistakes.

const allSessions: any[] = [];

Copilot AI Jan 3, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The 'allSessions' array is typed as 'any[]' which bypasses TypeScript's type checking. Consider defining a proper interface for session objects to improve type safety and code maintainability.

Copilot uses AI. Check for mistakes.

Copilot AI Jan 3, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The 'buildPipeline' function lacks documentation explaining its purpose, parameters, and the complex MongoDB aggregation logic it contains. Consider adding a JSDoc comment that describes the function's purpose, the meaning of the 'from' and 'to' parameters, and a brief explanation of the session detection algorithm (e.g., sessions are detected when there's a gap greater than sessionBreakThreshold between consecutive tracks).

Suggested change
/**
* Builds a MongoDB aggregation pipeline that finds listening sessions for a given
* time window. The pipeline filters tracks between the provided dates, orders
* them chronologically, and uses a gap-based algorithm to split them into
* sessions.
*
* A new session is started whenever the gap between the end of one track and
* the start of the next track is greater than `sessionBreakThreshold`
* (computed as the time between `$$value.last.played_at + durationMs` and
* `$$this.played_at`). Consecutive tracks with a gap less than or equal to the
* threshold are considered part of the same session.
*
* @param from - Start of the time window (inclusive) for which sessions are
* computed.
* @param to - End of the time window (exclusive or clamped to this boundary)
* for which sessions are computed.
* @returns A MongoDB aggregation pipeline array that, when run, returns the
* detected sessions and their associated tracks within the given window.
*/

Copilot uses AI. Check for mistakes.
const buildPipeline = (from: Date, to: Date): any[] => {

Copilot AI Jan 3, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The pipeline array type is declared as 'any[]' which bypasses TypeScript's type checking. Consider defining a proper type for the MongoDB aggregation pipeline stages to improve type safety and code maintainability.

Copilot uses AI. Check for mistakes.
const subtract = {
$cond: [
"$$value.last",
{
$subtract: [
"$$this.played_at",
{
$add: ["$$value.last.played_at", "$$value.last.durationMs"],
},
],
},
sessionBreakThreshold + 1,
],
};

const item = { subtract, info: "$$this" };

return [
...basicMatch(userId, from, to),
{ $sort: { played_at: 1 } },
{
$group: {
_id: "$owner",
infos: { $push: "$$ROOT" },
},
},
{
$addFields: {
distanceToLast: {
$reduce: {
input: "$infos",
initialValue: { distance: [], current: [], last: null },
in: {
distance: {
$concatArrays: [
"$$value.distance",
{
$cond: {
if: {
$gt: [subtract, sessionBreakThreshold],
},
then: ["$$value.current"],
else: [],
},
then: ["$$value.current"],
else: [],
},
},
],
},
current: {
$cond: {
if: {
$gt: [subtract, sessionBreakThreshold],
},
then: [item],
else: {
$concatArrays: ["$$value.current", [item]],
],
},
current: {
$cond: {
if: {
$gt: [subtract, sessionBreakThreshold],
},
then: [item],
else: {
$concatArrays: ["$$value.current", [item]],
},
},
},
last: "$$this",
},
last: "$$this",
},
},
},
},
},
{ $unset: ["infos", "distanceToLast.last"] },
{
$addFields: {
"distanceToLast.distance": {
$concatArrays: [
"$distanceToLast.distance",
["$distanceToLast.current"],
],
{ $unset: ["infos", "distanceToLast.last"] },
{
$addFields: {
"distanceToLast.distance": {
$concatArrays: [
"$distanceToLast.distance",
["$distanceToLast.current"],
],
},
},
},
},
{ $unset: "distanceToLast.current" },
{
$unwind: {
path: "$distanceToLast.distance",
{ $unset: "distanceToLast.current" },
{
$unwind: {
path: "$distanceToLast.distance",
},
},
},
{
$addFields: {
sessionLength: {
$subtract: [
{ $last: "$distanceToLast.distance.info.played_at" },
{ $first: "$distanceToLast.distance.info.played_at" },
],
{
$addFields: {
sessionLength: {
$subtract: [
{ $last: "$distanceToLast.distance.info.played_at" },
{ $first: "$distanceToLast.distance.info.played_at" },
],
},
},
},
},
{ $sort: { sessionLength: -1 } },
{ $limit: 5 },
{
$lookup: {
from: "tracks",
localField: "distanceToLast.distance.info.id",
foreignField: "id",
as: "full_tracks",
{ $sort: { sessionLength: -1 } },
{ $limit: 5 },
{
$lookup: {
from: "tracks",
localField: "distanceToLast.distance.info.id",
foreignField: "id",
as: "full_tracks",
},
},
},
]);
];
};

longestSessions.forEach(longestSession => {
longestSession.full_tracks = Object.fromEntries(
longestSession.full_tracks.map((track: any) => [track.id, track]),
);
});
for (const w of windows) {
const chunkSessions = await InfosModel.aggregate(buildPipeline(w.from, w.to));
chunkSessions.forEach((s: any) => {
s.full_tracks = Object.fromEntries(
s.full_tracks.map((t: any) => [t.id, t]),
);
allSessions.push(s);
});
}
Comment on lines +800 to +808

Copilot AI Jan 3, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The aggregation queries are executed sequentially in a loop, which means the total processing time will be the sum of all individual query times. Consider using Promise.all to execute all batch queries in parallel, which could significantly reduce the overall processing time especially for large date ranges spanning multiple years.

Suggested change
for (const w of windows) {
const chunkSessions = await InfosModel.aggregate(buildPipeline(w.from, w.to));
chunkSessions.forEach((s: any) => {
s.full_tracks = Object.fromEntries(
s.full_tracks.map((t: any) => [t.id, t]),
);
allSessions.push(s);
});
}
const sessionPromises = windows.map((w) =>
InfosModel.aggregate(buildPipeline(w.from, w.to)),
);
const chunkSessionsArrays = await Promise.all(sessionPromises);
chunkSessionsArrays.forEach((chunkSessions) => {
chunkSessions.forEach((s: any) => {
s.full_tracks = Object.fromEntries(
s.full_tracks.map((t: any) => [t.id, t]),
);
allSessions.push(s);
});
});

Copilot uses AI. Check for mistakes.

if (!allSessions.length) return [];

return longestSessions;
allSessions.sort((a, b) => b.sessionLength - a.sessionLength);
return allSessions.slice(0, 5);
Comment on lines +812 to +813

Copilot AI Jan 3, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The magic number '5' is hardcoded in two places (lines 788 and 813) representing the number of top sessions to return. Consider extracting this into a named constant at the function level (e.g., 'TOP_SESSIONS_LIMIT = 5') to improve maintainability and make it clear that these two values should be kept in sync.

Suggested change
allSessions.sort((a, b) => b.sessionLength - a.sessionLength);
return allSessions.slice(0, 5);
const TOP_SESSIONS_LIMIT = 5;
allSessions.sort((a, b) => b.sessionLength - a.sessionLength);
return allSessions.slice(0, TOP_SESSIONS_LIMIT);

Copilot uses AI. Check for mistakes.
};

export const getRankOf = async (
Expand Down Expand Up @@ -819,4 +849,4 @@ export const getRankOf = async (
},
]);
return res[0];
};
};