Tickets and task retrieval optimization

This commit is contained in:
Marc Rejohn Castillano 2026-03-01 18:27:19 +08:00
parent 294d3f7470
commit 2100516238
2 changed files with 273 additions and 117 deletions

View File

@ -105,54 +105,17 @@ class TaskQuery {
}
}
final tasksProvider = StreamProvider<List<Task>>((ref) {
final client = ref.watch(supabaseClientProvider);
final profileAsync = ref.watch(currentProfileProvider);
final ticketsAsync = ref.watch(ticketsProvider);
final assignmentsAsync = ref.watch(userOfficesProvider);
final query = ref.watch(tasksQueryProvider);
final profile = profileAsync.valueOrNull;
if (profile == null) {
return Stream.value(const <Task>[]);
}
final isGlobal =
profile.role == 'admin' ||
profile.role == 'dispatcher' ||
profile.role == 'it_staff';
// For RBAC early-exit: if the user has no accessible tickets/offices,
// avoid subscribing to the full tasks stream.
List<String> earlyAllowedTicketIds =
ticketsAsync.valueOrNull?.map((ticket) => ticket.id).toList() ??
<String>[];
List<String> earlyOfficeIds =
assignmentsAsync.valueOrNull
?.where((assignment) => assignment.userId == profile.id)
.map((assignment) => assignment.officeId)
.toSet()
.toList() ??
<String>[];
if (!isGlobal && earlyAllowedTicketIds.isEmpty && earlyOfficeIds.isEmpty) {
return Stream.value(const <Task>[]);
}
// Wrap realtime stream with recovery logic
final wrapper = StreamRecoveryWrapper<Task>(
stream: client.from('tasks').stream(primaryKey: ['id']),
onPollData: () async {
final data = await client.from('tasks').select();
return data.cast<Map<String, dynamic>>().map(Task.fromMap).toList();
},
fromMap: Task.fromMap,
);
ref.onDispose(wrapper.dispose);
// Process tasks with filtering/pagination after recovery
return wrapper.stream.asyncMap((result) async {
final rowsList = result.data
/// Builds the isolate payload from a list of [Task] objects and the current
/// query/access context. Extracted so the initial REST seed and the realtime
/// stream listener can share the same logic without duplication.
Map<String, dynamic> _buildTaskPayload({
required List<Task> tasks,
required bool isGlobal,
required List<String> allowedTicketIds,
required List<String> allowedOfficeIds,
required TaskQuery query,
}) {
final rowsList = tasks
.map(
(task) => <String, dynamic>{
'id': task.id,
@ -181,17 +144,7 @@ final tasksProvider = StreamProvider<List<Task>>((ref) {
)
.toList();
final allowedTicketIds =
ticketsAsync.valueOrNull?.map((ticket) => ticket.id).toList() ??
<String>[];
final allowedOfficeIds =
assignmentsAsync.valueOrNull
?.where((assignment) => assignment.userId == profile.id)
.map((assignment) => assignment.officeId)
.toList() ??
<String>[];
final payload = <String, dynamic>{
return {
'rows': rowsList,
'isGlobal': isGlobal,
'allowedTicketIds': allowedTicketIds,
@ -203,17 +156,138 @@ final tasksProvider = StreamProvider<List<Task>>((ref) {
'dateStart': query.dateRange?.start.millisecondsSinceEpoch,
'dateEnd': query.dateRange?.end.millisecondsSinceEpoch,
};
}
final tasksProvider = StreamProvider<List<Task>>((ref) {
final client = ref.watch(supabaseClientProvider);
final profileAsync = ref.watch(currentProfileProvider);
final ticketsAsync = ref.watch(ticketsProvider);
final assignmentsAsync = ref.watch(userOfficesProvider);
final query = ref.watch(tasksQueryProvider);
final profile = profileAsync.valueOrNull;
if (profile == null) {
return Stream.value(const <Task>[]);
}
final isGlobal =
profile.role == 'admin' ||
profile.role == 'dispatcher' ||
profile.role == 'it_staff';
final allowedTicketIds =
ticketsAsync.valueOrNull?.map((t) => t.id).toList() ?? <String>[];
final allowedOfficeIds =
assignmentsAsync.valueOrNull
?.where((a) => a.userId == profile.id)
.map((a) => a.officeId)
.toSet()
.toList() ??
<String>[];
// For non-global users with no assigned offices/tickets, skip subscribing.
if (!isGlobal && allowedTicketIds.isEmpty && allowedOfficeIds.isEmpty) {
return Stream.value(const <Task>[]);
}
// Wrap realtime stream with recovery logic
final wrapper = StreamRecoveryWrapper<Task>(
stream: client.from('tasks').stream(primaryKey: ['id']),
onPollData: () async {
final data = await client.from('tasks').select();
return data.cast<Map<String, dynamic>>().map(Task.fromMap).toList();
},
fromMap: Task.fromMap,
);
ref.onDispose(wrapper.dispose);
var lastResultHash = '';
Timer? debounceTimer;
// broadcast() so Riverpod and any other listener can both receive events.
final controller = StreamController<List<Task>>.broadcast();
void emitDebounced(List<Task> tasks) {
debounceTimer?.cancel();
debounceTimer = Timer(const Duration(milliseconds: 150), () {
if (!controller.isClosed) controller.add(tasks);
});
}
ref.onDispose(() {
debounceTimer?.cancel();
controller.close();
});
// Immediate REST seed
// Fire a one-shot HTTP fetch right now so the UI can render before the
// WebSocket realtime channel is fully established. Eliminates loading delay
// on web and initial flash on mobile. Hash check prevents a duplicate
// rebuild if both the seed and the realtime stream arrive with the same data.
unawaited(
Future(() async {
try {
final data = await client.from('tasks').select();
final raw = data
.cast<Map<String, dynamic>>()
.map(Task.fromMap)
.toList();
final payload = _buildTaskPayload(
tasks: raw,
isGlobal: isGlobal,
allowedTicketIds: allowedTicketIds,
allowedOfficeIds: allowedOfficeIds,
query: query,
);
final processed = await compute(_processTasksInIsolate, payload);
final tasks = (processed as List<dynamic>)
.cast<Map<String, dynamic>>()
.map(Task.fromMap)
.toList();
final hash = tasks.fold('', (h, t) => '$h${t.id}');
if (!controller.isClosed && hash != lastResultHash) {
lastResultHash = hash;
controller.add(tasks); // emit immediately no debounce
}
} catch (e) {
debugPrint('[tasksProvider] initial seed error: $e');
}
}),
);
debugPrint('[tasksProvider] processed ${tasks.length} tasks');
return tasks;
});
// Realtime stream
// Processes every realtime event through the same isolate. Debounced so
// rapid consecutive events (e.g. bulk inserts) don't cause repeated renders.
wrapper.stream
.asyncMap((result) async {
final payload = _buildTaskPayload(
tasks: result.data,
isGlobal: isGlobal,
allowedTicketIds: allowedTicketIds,
allowedOfficeIds: allowedOfficeIds,
query: query,
);
final processed = await compute(_processTasksInIsolate, payload);
return (processed as List<dynamic>)
.cast<Map<String, dynamic>>()
.map(Task.fromMap)
.toList();
})
.listen(
(tasks) {
final hash = tasks.fold('', (h, t) => '$h${t.id}');
if (hash != lastResultHash) {
lastResultHash = hash;
emitDebounced(tasks);
}
},
onError: (Object e) {
debugPrint('[tasksProvider] stream error: $e');
controller.addError(e);
},
);
return controller.stream;
});
// Runs inside a background isolate to filter/sort tasks represented as

View File

@ -120,38 +120,16 @@ class TicketQuery {
}
}
final ticketsProvider = StreamProvider<List<Ticket>>((ref) {
final client = ref.watch(supabaseClientProvider);
final profileAsync = ref.watch(currentProfileProvider);
final assignmentsAsync = ref.watch(userOfficesProvider);
final query = ref.watch(ticketsQueryProvider);
final profile = profileAsync.valueOrNull;
if (profile == null) {
return Stream.value(const <Ticket>[]);
}
final isGlobal =
profile.role == 'admin' ||
profile.role == 'dispatcher' ||
profile.role == 'it_staff';
// Wrap realtime stream with recovery logic
final wrapper = StreamRecoveryWrapper<Ticket>(
stream: client.from('tickets').stream(primaryKey: ['id']),
onPollData: () async {
// Polling fallback: fetch all tickets once
final data = await client.from('tickets').select();
return data.cast<Map<String, dynamic>>().map(Ticket.fromMap).toList();
},
fromMap: Ticket.fromMap,
);
ref.onDispose(wrapper.dispose);
// Process tickets with filtering/pagination after recovery
return wrapper.stream.asyncMap((result) async {
final rowsList = result.data
/// Builds the isolate payload from a list of [Ticket] objects and the current
/// query/access context. Extracted so the initial REST seed and the realtime
/// stream listener can share the same logic without duplication.
Map<String, dynamic> _buildTicketPayload({
required List<Ticket> tickets,
required bool isGlobal,
required List<String> allowedOfficeIds,
required TicketQuery query,
}) {
final rowsList = tickets
.map(
(ticket) => <String, dynamic>{
'id': ticket.id,
@ -168,14 +146,7 @@ final ticketsProvider = StreamProvider<List<Ticket>>((ref) {
)
.toList();
final allowedOfficeIds =
assignmentsAsync.valueOrNull
?.where((assignment) => assignment.userId == profile.id)
.map((assignment) => assignment.officeId)
.toList() ??
<String>[];
final payload = <String, dynamic>{
return {
'rows': rowsList,
'isGlobal': isGlobal,
'allowedOfficeIds': allowedOfficeIds,
@ -187,17 +158,128 @@ final ticketsProvider = StreamProvider<List<Ticket>>((ref) {
'dateStart': query.dateRange?.start.millisecondsSinceEpoch,
'dateEnd': query.dateRange?.end.millisecondsSinceEpoch,
};
}
final ticketsProvider = StreamProvider<List<Ticket>>((ref) {
final client = ref.watch(supabaseClientProvider);
final profileAsync = ref.watch(currentProfileProvider);
final assignmentsAsync = ref.watch(userOfficesProvider);
final query = ref.watch(ticketsQueryProvider);
final profile = profileAsync.valueOrNull;
if (profile == null) {
return Stream.value(const <Ticket>[]);
}
final isGlobal =
profile.role == 'admin' ||
profile.role == 'dispatcher' ||
profile.role == 'it_staff';
final allowedOfficeIds =
assignmentsAsync.valueOrNull
?.where((a) => a.userId == profile.id)
.map((a) => a.officeId)
.toList() ??
<String>[];
// Wrap realtime stream with recovery logic
final wrapper = StreamRecoveryWrapper<Ticket>(
stream: client.from('tickets').stream(primaryKey: ['id']),
onPollData: () async {
final data = await client.from('tickets').select();
return data.cast<Map<String, dynamic>>().map(Ticket.fromMap).toList();
},
fromMap: Ticket.fromMap,
);
ref.onDispose(wrapper.dispose);
var lastResultHash = '';
Timer? debounceTimer;
// broadcast() so Riverpod and any other listener can both receive events.
final controller = StreamController<List<Ticket>>.broadcast();
void emitDebounced(List<Ticket> tickets) {
debounceTimer?.cancel();
debounceTimer = Timer(const Duration(milliseconds: 150), () {
if (!controller.isClosed) controller.add(tickets);
});
}
ref.onDispose(() {
debounceTimer?.cancel();
controller.close();
});
// Immediate REST seed
// Fire a one-shot HTTP fetch right now so the UI can render before the
// WebSocket realtime channel is fully established. This eliminates the
// loading delay on web (WebSocket ~200-500 ms) and the initial flash on
// mobile. The realtime stream takes over afterwards; the hash check below
// prevents a duplicate rebuild if both arrive with identical data.
unawaited(
Future(() async {
try {
final data = await client.from('tickets').select();
final raw = data
.cast<Map<String, dynamic>>()
.map(Ticket.fromMap)
.toList();
final payload = _buildTicketPayload(
tickets: raw,
isGlobal: isGlobal,
allowedOfficeIds: allowedOfficeIds,
query: query,
);
final processed = await compute(_processTicketsInIsolate, payload);
final tickets = (processed as List<dynamic>)
.cast<Map<String, dynamic>>()
.map(Ticket.fromMap)
.toList();
final hash = tickets.fold('', (h, t) => '$h${t.id}');
if (!controller.isClosed && hash != lastResultHash) {
lastResultHash = hash;
controller.add(tickets); // emit immediately no debounce
}
} catch (e) {
debugPrint('[ticketsProvider] initial seed error: $e');
}
}),
);
debugPrint('[ticketsProvider] processed ${tickets.length} tickets');
return tickets;
});
// Realtime stream
// Processes every realtime event through the same isolate. Debounced so
// rapid consecutive events (e.g. bulk inserts) don't cause repeated renders.
wrapper.stream
.asyncMap((result) async {
final payload = _buildTicketPayload(
tickets: result.data,
isGlobal: isGlobal,
allowedOfficeIds: allowedOfficeIds,
query: query,
);
final processed = await compute(_processTicketsInIsolate, payload);
return (processed as List<dynamic>)
.cast<Map<String, dynamic>>()
.map(Ticket.fromMap)
.toList();
})
.listen(
(tickets) {
final hash = tickets.fold('', (h, t) => '$h${t.id}');
if (hash != lastResultHash) {
lastResultHash = hash;
emitDebounced(tickets);
}
},
onError: (Object e) {
debugPrint('[ticketsProvider] stream error: $e');
controller.addError(e);
},
);
return controller.stream;
});
// Runs inside a background isolate. Accepts a serializable payload and