zulip/web/src/stream_topic_history.ts

346 lines
11 KiB
TypeScript
Raw Normal View History

import assert from "minimalistic-assert";
import {all_messages_data} from "./all_messages_data";
import {FoldDict} from "./fold_dict";
import * as message_util from "./message_util";
import * as sub_store from "./sub_store";
import type {StreamSubscription} from "./sub_store";
import * as unread from "./unread";
// stream_id -> PerStreamHistory object
const stream_dict = new Map<number, PerStreamHistory>();
const fetched_stream_ids = new Set<number>();
const request_pending_stream_ids = new Set<number>();
// This is stream_topic_history_util.get_server_history.
// We have to indirectly set it to avoid a circular dependency.
let update_topic_last_message_id: (stream_id: number, topic_name: string) => void;
export function set_update_topic_last_message_id(
f: (stream_id: number, topic_name: string) => void,
): void {
update_topic_last_message_id = f;
}
export function all_topics_in_cache(sub: StreamSubscription): boolean {
// Checks whether this browser's cache of contiguous messages
// (used to locally render narrows) in all_messages_data has all
// messages from a given stream. Because all_messages_data is a range,
// we just need to compare it to the range of history on the stream.
// If the cache isn't initialized, it's a clear false.
if (all_messages_data === undefined || all_messages_data.empty()) {
return false;
}
// If the cache doesn't have the latest messages, we can't be sure
// we have all topics.
if (!all_messages_data.fetch_status.has_found_newest()) {
return false;
}
if (sub.first_message_id === null) {
// If the stream has no message history, we have it all
// vacuously. This should be a very rare condition, since
// stream creation sends a message.
return true;
}
// Now, we can just compare the first cached message to the first
// message ID in the stream; if it's older, we're good, otherwise,
// we might be missing the oldest topics in this stream in our
// cache.
const first_cached_message = all_messages_data.first();
return first_cached_message!.id <= sub.first_message_id;
}
export function is_complete_for_stream_id(stream_id: number): boolean {
if (fetched_stream_ids.has(stream_id)) {
return true;
}
const sub = sub_store.get(stream_id);
const in_cache = sub !== undefined && all_topics_in_cache(sub);
if (in_cache) {
/*
If the stream is cached, we can add it to
fetched_stream_ids. Note that for the opposite
scenario, we don't delete from
fetched_stream_ids, because we may just be
waiting for the initial message fetch.
*/
fetched_stream_ids.add(stream_id);
}
return in_cache;
}
export function stream_has_topics(stream_id: number): boolean {
if (!stream_dict.has(stream_id)) {
return false;
}
const history = stream_dict.get(stream_id);
assert(history !== undefined);
return history.has_topics();
}
export type TopicHistoryEntry = {
count: number;
message_id: number;
pretty_name: string;
};
type ServerTopicHistoryEntry = {
name: string;
max_id: number;
};
export class PerStreamHistory {
/*
For a given stream, this structure has a dictionary of topics.
The main getter of this object is get_recent_topic_names, and
we just sort on the fly every time we are called.
Attributes for a topic are:
* message_id: The latest message_id in the topic. This is to
the best of our knowledge, and may not be accurate if
we have not seen all the messages in the topic.
* pretty_name: The topic_name, with original case.
* count: Number of known messages in the topic. Used to detect
when the last messages in a topic were moved to other topics or
deleted.
*/
topics = new FoldDict<TopicHistoryEntry>();
// Most recent message ID for the stream.
max_message_id = 0;
stream_id: number;
constructor(stream_id: number) {
this.stream_id = stream_id;
}
has_topics(): boolean {
return this.topics.size !== 0;
}
update_stream_max_message_id(message_id: number): void {
if (message_id > this.max_message_id) {
this.max_message_id = message_id;
}
}
add_or_update(topic_name: string, message_id: number): void {
// The `message_id` provided here can easily be far from the latest
// message in the topic, but it is more important for us to cache the topic
// for autocomplete purposes than to have an accurate max message ID.
this.update_stream_max_message_id(message_id);
const existing = this.topics.get(topic_name);
if (!existing) {
this.topics.set(topic_name, {
message_id,
pretty_name: topic_name,
count: 1,
});
return;
}
existing.count += 1;
2017-07-27 12:57:37 +02:00
if (message_id > existing.message_id) {
existing.message_id = message_id;
existing.pretty_name = topic_name;
}
}
maybe_remove(topic_name: string, num_messages: number): void {
const existing = this.topics.get(topic_name);
if (!existing || existing.count === 0) {
return;
}
if (existing.count <= num_messages) {
this.topics.delete(topic_name);
if (!is_complete_for_stream_id(this.stream_id)) {
// Request server for latest message in topic if we
// cannot be sure that we have all messages in the topic.
update_topic_last_message_id(this.stream_id, topic_name);
return;
}
}
existing.count -= num_messages;
}
add_history(server_history: ServerTopicHistoryEntry[]): void {
// This method populates list of topics from the server.
2017-07-27 12:57:37 +02:00
js: Automatically convert _.each to for…of. This commit was automatically generated by the following script, followed by lint --fix and a few small manual lint-related cleanups. import * as babelParser from "recast/parsers/babel"; import * as recast from "recast"; import * as tsParser from "recast/parsers/typescript"; import { builders as b, namedTypes as n } from "ast-types"; import { Context } from "ast-types/lib/path-visitor"; import K from "ast-types/gen/kinds"; import { NodePath } from "ast-types/lib/node-path"; import assert from "assert"; import fs from "fs"; import path from "path"; import process from "process"; const checkExpression = (node: n.Node): node is K.ExpressionKind => n.Expression.check(node); const checkStatement = (node: n.Node): node is K.StatementKind => n.Statement.check(node); for (const file of process.argv.slice(2)) { console.log("Parsing", file); const ast = recast.parse(fs.readFileSync(file, { encoding: "utf8" }), { parser: path.extname(file) === ".ts" ? tsParser : babelParser, }); let changed = false; let inLoop = false; let replaceReturn = false; const visitLoop = (...args: string[]) => function(this: Context, path: NodePath) { for (const arg of args) { this.visit(path.get(arg)); } const old = { inLoop }; inLoop = true; this.visit(path.get("body")); inLoop = old.inLoop; return false; }; recast.visit(ast, { visitDoWhileStatement: visitLoop("test"), visitExpressionStatement(path) { const { expression, comments } = path.node; let valueOnly; if ( n.CallExpression.check(expression) && n.MemberExpression.check(expression.callee) && !expression.callee.computed && n.Identifier.check(expression.callee.object) && expression.callee.object.name === "_" && n.Identifier.check(expression.callee.property) && ["each", "forEach"].includes(expression.callee.property.name) && [2, 3].includes(expression.arguments.length) && checkExpression(expression.arguments[0]) && (n.FunctionExpression.check(expression.arguments[1]) || n.ArrowFunctionExpression.check(expression.arguments[1])) && [1, 2].includes(expression.arguments[1].params.length) && n.Identifier.check(expression.arguments[1].params[0]) && ((valueOnly = expression.arguments[1].params[1] === undefined) || n.Identifier.check(expression.arguments[1].params[1])) && (expression.arguments[2] === undefined || n.ThisExpression.check(expression.arguments[2])) ) { const old = { inLoop, replaceReturn }; inLoop = false; replaceReturn = true; this.visit( path .get("expression") .get("arguments") .get(1) .get("body") ); inLoop = old.inLoop; replaceReturn = old.replaceReturn; const [right, { body, params }] = expression.arguments; const loop = b.forOfStatement( b.variableDeclaration("let", [ b.variableDeclarator( valueOnly ? params[0] : b.arrayPattern([params[1], params[0]]) ), ]), valueOnly ? right : b.callExpression( b.memberExpression(right, b.identifier("entries")), [] ), checkStatement(body) ? body : b.expressionStatement(body) ); loop.comments = comments; path.replace(loop); changed = true; } this.traverse(path); }, visitForStatement: visitLoop("init", "test", "update"), visitForInStatement: visitLoop("left", "right"), visitForOfStatement: visitLoop("left", "right"), visitFunction(path) { this.visit(path.get("params")); const old = { replaceReturn }; replaceReturn = false; this.visit(path.get("body")); replaceReturn = old.replaceReturn; return false; }, visitReturnStatement(path) { if (replaceReturn) { assert(!inLoop); // could use labeled continue if this ever fires const { argument, comments } = path.node; if (argument === null) { const s = b.continueStatement(); s.comments = comments; path.replace(s); } else { const s = b.expressionStatement(argument); s.comments = comments; path.replace(s, b.continueStatement()); } return false; } this.traverse(path); }, visitWhileStatement: visitLoop("test"), }); if (changed) { console.log("Writing", file); fs.writeFileSync(file, recast.print(ast).code, { encoding: "utf8" }); } } Signed-off-by: Anders Kaseorg <anders@zulipchat.com>
2020-02-06 06:19:47 +01:00
for (const obj of server_history) {
const topic_name = obj.name;
const message_id = obj.max_id;
2017-07-27 12:57:37 +02:00
const existing = this.topics.get(topic_name);
2017-07-27 12:57:37 +02:00
if (existing) {
// Trust out local data more, since it
// maintains counts.
continue;
2017-07-27 12:57:37 +02:00
}
// If we get here, we are either finding out about
// the topic for the first time, or we are getting
// more current data for it.
this.topics.set(topic_name, {
message_id,
pretty_name: topic_name,
count: 0,
2017-07-27 12:57:37 +02:00
});
this.update_stream_max_message_id(message_id);
js: Automatically convert _.each to for…of. This commit was automatically generated by the following script, followed by lint --fix and a few small manual lint-related cleanups. import * as babelParser from "recast/parsers/babel"; import * as recast from "recast"; import * as tsParser from "recast/parsers/typescript"; import { builders as b, namedTypes as n } from "ast-types"; import { Context } from "ast-types/lib/path-visitor"; import K from "ast-types/gen/kinds"; import { NodePath } from "ast-types/lib/node-path"; import assert from "assert"; import fs from "fs"; import path from "path"; import process from "process"; const checkExpression = (node: n.Node): node is K.ExpressionKind => n.Expression.check(node); const checkStatement = (node: n.Node): node is K.StatementKind => n.Statement.check(node); for (const file of process.argv.slice(2)) { console.log("Parsing", file); const ast = recast.parse(fs.readFileSync(file, { encoding: "utf8" }), { parser: path.extname(file) === ".ts" ? tsParser : babelParser, }); let changed = false; let inLoop = false; let replaceReturn = false; const visitLoop = (...args: string[]) => function(this: Context, path: NodePath) { for (const arg of args) { this.visit(path.get(arg)); } const old = { inLoop }; inLoop = true; this.visit(path.get("body")); inLoop = old.inLoop; return false; }; recast.visit(ast, { visitDoWhileStatement: visitLoop("test"), visitExpressionStatement(path) { const { expression, comments } = path.node; let valueOnly; if ( n.CallExpression.check(expression) && n.MemberExpression.check(expression.callee) && !expression.callee.computed && n.Identifier.check(expression.callee.object) && expression.callee.object.name === "_" && n.Identifier.check(expression.callee.property) && ["each", "forEach"].includes(expression.callee.property.name) && [2, 3].includes(expression.arguments.length) && checkExpression(expression.arguments[0]) && (n.FunctionExpression.check(expression.arguments[1]) || n.ArrowFunctionExpression.check(expression.arguments[1])) && [1, 2].includes(expression.arguments[1].params.length) && n.Identifier.check(expression.arguments[1].params[0]) && ((valueOnly = expression.arguments[1].params[1] === undefined) || n.Identifier.check(expression.arguments[1].params[1])) && (expression.arguments[2] === undefined || n.ThisExpression.check(expression.arguments[2])) ) { const old = { inLoop, replaceReturn }; inLoop = false; replaceReturn = true; this.visit( path .get("expression") .get("arguments") .get(1) .get("body") ); inLoop = old.inLoop; replaceReturn = old.replaceReturn; const [right, { body, params }] = expression.arguments; const loop = b.forOfStatement( b.variableDeclaration("let", [ b.variableDeclarator( valueOnly ? params[0] : b.arrayPattern([params[1], params[0]]) ), ]), valueOnly ? right : b.callExpression( b.memberExpression(right, b.identifier("entries")), [] ), checkStatement(body) ? body : b.expressionStatement(body) ); loop.comments = comments; path.replace(loop); changed = true; } this.traverse(path); }, visitForStatement: visitLoop("init", "test", "update"), visitForInStatement: visitLoop("left", "right"), visitForOfStatement: visitLoop("left", "right"), visitFunction(path) { this.visit(path.get("params")); const old = { replaceReturn }; replaceReturn = false; this.visit(path.get("body")); replaceReturn = old.replaceReturn; return false; }, visitReturnStatement(path) { if (replaceReturn) { assert(!inLoop); // could use labeled continue if this ever fires const { argument, comments } = path.node; if (argument === null) { const s = b.continueStatement(); s.comments = comments; path.replace(s); } else { const s = b.expressionStatement(argument); s.comments = comments; path.replace(s, b.continueStatement()); } return false; } this.traverse(path); }, visitWhileStatement: visitLoop("test"), }); if (changed) { console.log("Writing", file); fs.writeFileSync(file, recast.print(ast).code, { encoding: "utf8" }); } } Signed-off-by: Anders Kaseorg <anders@zulipchat.com>
2020-02-06 06:19:47 +01:00
}
}
2017-07-27 12:57:37 +02:00
get_recent_topic_names(): string[] {
const my_recents = [...this.topics.values()];
/* Add any older topics with unreads that may not be present
* in our local cache. */
const missing_topics = unread.get_missing_topics({
stream_id: this.stream_id,
topic_dict: this.topics,
});
const recents = [...my_recents, ...missing_topics];
recents.sort((a, b) => b.message_id - a.message_id);
const names = recents.map((obj) => obj.pretty_name);
return names;
}
get_max_message_id(): number {
return this.max_message_id;
}
}
export function remove_messages(opts: {
stream_id: number;
topic_name: string;
num_messages: number;
max_removed_msg_id: number;
}): void {
const stream_id = opts.stream_id;
const topic_name = opts.topic_name;
const num_messages = opts.num_messages;
const max_removed_msg_id = opts.max_removed_msg_id;
const history = stream_dict.get(stream_id);
// This is the special case of "removing" a message from
// a topic, which happens when we edit topics.
if (!history) {
return;
}
// Adjust our local data structures to account for the
// removal of messages from a topic. We can also remove
// the topic if it has no messages left or if we cannot
// locally determine the current state of the topic.
// So, it is important that we return below if we don't have
// the topic cached.
history.maybe_remove(topic_name, num_messages);
const existing_topic = history.topics.get(topic_name);
if (!existing_topic) {
return;
}
// Update max_message_id in topic
if (existing_topic.message_id <= max_removed_msg_id) {
const msgs_in_topic = message_util.get_messages_in_topic(stream_id, topic_name);
let max_message_id = 0;
for (const msg of msgs_in_topic) {
if (msg.id > max_message_id) {
max_message_id = msg.id;
}
}
existing_topic.message_id = max_message_id;
}
// Update max_message_id in stream
if (history.max_message_id <= max_removed_msg_id) {
history.max_message_id = message_util.get_max_message_id_in_stream(stream_id);
}
}
export function find_or_create(stream_id: number): PerStreamHistory {
let history = stream_dict.get(stream_id);
if (!history) {
history = new PerStreamHistory(stream_id);
stream_dict.set(stream_id, history);
}
return history;
}
export function add_message(opts: {
stream_id: number;
message_id: number;
topic_name: string;
}): void {
const stream_id = opts.stream_id;
const message_id = opts.message_id;
const topic_name = opts.topic_name;
const history = find_or_create(stream_id);
history.add_or_update(topic_name, message_id);
}
export function add_history(stream_id: number, server_history: ServerTopicHistoryEntry[]): void {
const history = find_or_create(stream_id);
2017-07-27 12:57:37 +02:00
history.add_history(server_history);
fetched_stream_ids.add(stream_id);
}
2017-07-27 12:57:37 +02:00
export function has_history_for(stream_id: number): boolean {
return fetched_stream_ids.has(stream_id);
}
export function get_recent_topic_names(stream_id: number): string[] {
const history = find_or_create(stream_id);
return history.get_recent_topic_names();
}
export function get_max_message_id(stream_id: number): number {
const history = find_or_create(stream_id);
return history.get_max_message_id();
}
export function reset(): void {
// This is only used by tests.
stream_dict.clear();
fetched_stream_ids.clear();
request_pending_stream_ids.clear();
}
export function is_request_pending_for(stream_id: number): boolean {
return request_pending_stream_ids.has(stream_id);
}
export function add_request_pending_for(stream_id: number): void {
request_pending_stream_ids.add(stream_id);
}
export function remove_request_pending_for(stream_id: number): void {
request_pending_stream_ids.delete(stream_id);
}