feat: init

This commit is contained in:
montaghanmy
2023-03-23 11:03:16 +01:00
commit 10fe6f78d1
11518 changed files with 509786 additions and 0 deletions
@@ -0,0 +1 @@
declare module "emoji-name-map";
+21
View File
@@ -0,0 +1,21 @@
import { FastifyReply, FastifyRequest } from "fastify";
import SocketIO from "socket.io";
import { User } from "../utils/types";
export interface authenticateDecorator {
(request: FastifyRequest): FastifyRequest;
(reply: FastifyReply): FastifyReply;
}
declare module "fastify" {
interface FastifyInstance {
phpnodeAuthenticate(): void;
authenticate(): void;
authenticateOptional(): void;
io: SocketIO.Server;
}
interface FastifyRequest {
currentUser: User;
}
}
@@ -0,0 +1 @@
declare module "free-email-domains";
@@ -0,0 +1 @@
declare module "get-website-favicon";
@@ -0,0 +1 @@
declare module "multistream";
+1
View File
@@ -0,0 +1 @@
declare module "unoconv-promise";
@@ -0,0 +1 @@
declare module "uuid-time";
@@ -0,0 +1,14 @@
import { CommandModule } from "yargs";
const command: CommandModule = {
describe: "Manage Twake Applications",
command: "applications <command>",
builder: yargs =>
yargs.commandDir("applications_cmds", {
visit: commandModule => commandModule.default,
}),
// eslint-disable-next-line @typescript-eslint/no-empty-function
handler: () => {},
};
export default command;
@@ -0,0 +1,54 @@
import yargs from "yargs";
import ora from "ora";
import twake from "../../../twake";
import Table from "cli-table";
import * as process from "process";
import gr from "../../../services/global-resolver";
/**
* Merge command parameters. Check the builder definition below for more details.
*/
type CLIArgs = Record<string, unknown>;
const services = [
"storage",
"counter",
"message-queue",
"platform-services",
"applications",
"auth",
"realtime",
"websocket",
// "user",
"search",
"database",
"webserver",
// "statistics",
];
const command: yargs.CommandModule<unknown, CLIArgs> = {
command: "ls",
describe: "command that allow you to list applications (unpublished only)",
handler: async _argv => {
const spinner = ora({ text: "Retrieving applications" }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
//
const unpublished = await gr.services.applications.marketplaceApps.listUnpublished(undefined);
//
const table = new Table({
head: ["ID", "Name", "Description"],
colWidths: [40, 20, 40],
});
unpublished.forEach((app: any) => {
table.push([app.id, app.identity.name, app.identity.description]);
});
spinner.stop();
console.log(table.toString());
await platform.stop();
process.exit(0);
},
};
export default command;
@@ -0,0 +1,89 @@
import yargs from "yargs";
import ora from "ora";
import twake from "../../../twake";
import Table from "cli-table";
import * as process from "process";
import gr from "../../../services/global-resolver";
/**
* Merge command parameters. Check the builder definition below for more details.
*/
type CLIArgs = {
id: string;
};
const services = [
"storage",
"counter",
"message-queue",
"platform-services",
"applications",
"auth",
"realtime",
"websocket",
"user",
"search",
"database",
"webserver",
"statistics",
];
const command: yargs.CommandModule<unknown, CLIArgs> = {
command: "publish <id>",
describe:
"command that allow you to validate an application and make it available on the marketplace",
builder: {
force: {
default: false,
type: "boolean",
description: "Force update unrequested application",
},
},
handler: async argv => {
if (argv.id) {
let spinner = ora({ text: "Retrieving application" }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
let app = await gr.services.applications.marketplaceApps.get({ id: argv.id }, undefined);
spinner.stop();
if (!app) {
console.error(`Application ${argv.id} not found`);
process.exit(1);
}
let table = new Table();
table.push(app);
table.push({ name: app.identity.name });
table.push({ description: app.identity.description });
table.push({ website: app.identity.website });
table.push({ requested: app.publication.requested });
table.push({ published: app.publication.published });
console.log(table.toString());
if (app.publication.published) {
console.error("Application already published");
process.exit(1);
}
if (!app.publication.requested && !argv.force) {
console.error(
"Application is not requested to be published. Use --force if you want to publish it anyway",
);
process.exit(1);
}
spinner = ora({ text: "Publishing application" }).start();
await gr.services.applications.marketplaceApps.publish({ id: argv.id }, undefined);
app = await gr.services.applications.marketplaceApps.get({ id: argv.id }, undefined);
spinner.stop();
console.log("Application published");
table = new Table();
table.push(app);
table.push({ published: app.publication.published });
console.log(table.toString());
}
process.exit(0);
},
};
export default command;
@@ -0,0 +1,73 @@
import yargs from "yargs";
import ora from "ora";
import twake from "../../../twake";
import Table from "cli-table";
import * as process from "process";
import gr from "../../../services/global-resolver";
/**
* Merge command parameters. Check the builder definition below for more details.
*/
type CLIArgs = {
id: string;
};
const services = [
"storage",
"counter",
"message-queue",
"platform-services",
"applications",
"auth",
"realtime",
"websocket",
"user",
"search",
"database",
"webserver",
"statistics",
];
const command: yargs.CommandModule<unknown, CLIArgs> = {
command: "unpublish <id>",
describe: "command that allow you to make application unavailable on the marketplace",
handler: async argv => {
if (argv.id) {
let spinner = ora({ text: "Retrieving application" }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
let app = await gr.services.applications.marketplaceApps.get({ id: argv.id }, undefined);
spinner.stop();
if (!app) {
console.error(`Application ${argv.id} not found`);
process.exit(1);
}
let table = new Table();
table.push(app);
table.push({ name: app.identity.name });
table.push({ description: app.identity.description });
table.push({ website: app.identity.website });
table.push({ requested: app.publication.requested });
table.push({ published: app.publication.published });
console.log(table.toString());
if (!app.publication.published) {
console.error("Application is not published");
process.exit(1);
}
spinner = ora({ text: "Unpublishing application" }).start();
await gr.services.applications.marketplaceApps.unpublish({ id: argv.id }, undefined);
app = await gr.services.applications.marketplaceApps.get({ id: argv.id }, undefined);
spinner.stop();
console.log("Application unpublished");
table = new Table();
table.push(app);
table.push({ published: app.publication.published });
console.log(table.toString());
}
process.exit(0);
},
};
export default command;
@@ -0,0 +1,14 @@
import { CommandModule } from "yargs";
const command: CommandModule = {
describe: "Manage Twake Console",
command: "console <command>",
builder: yargs =>
yargs.commandDir("console_cmds", {
visit: commandModule => commandModule.default,
}),
// eslint-disable-next-line @typescript-eslint/no-empty-function
handler: () => {},
};
export default command;
@@ -0,0 +1,257 @@
import yargs from "yargs";
import ora from "ora";
import twake from "../../../twake";
import { CompanyReport, UserReport } from "../../../services/console/types";
import Company from "../../../services/user/entities/company";
import gr from "../../../services/global-resolver";
import { ConsoleServiceImpl } from "../../../services/console/service";
/**
* Merge command parameters. Check the builder definition below for more details.
*/
type MergeParams = {
url: string;
concurrent: number;
dry: boolean;
console: string;
link: boolean;
csv: boolean;
client: string;
secret: string;
};
const services = [
"platform-services",
"user",
"channels",
"notifications",
"database",
"webserver",
"message-queue",
"console",
];
const command: yargs.CommandModule<MergeParams, MergeParams> = {
command: "merge",
describe: "Merge Twake Chat users in the Twake Console",
builder: {
url: {
default: "http://localhost:8080",
type: "string",
description: "URL of the Twake console",
},
concurrent: {
default: 1,
type: "number",
description: "Number of concurrent imports",
},
dry: {
default: false,
type: "boolean",
description: "Make a dry run without creating anything on the Twake console",
},
csv: {
default: false,
type: "boolean",
description: "Generate result as CSV",
},
console: {
default: "console",
type: "string",
description: "The console service identifier to user to link user and companies",
},
link: {
default: true,
type: "boolean",
description:
"Link the companies/users to external companies/user. Works with the --console parameter",
implies: "console",
},
client: {
default: "twake",
type: "string",
description: "Client identifier to be used to authenticate calls to the Console",
},
secret: {
default: "secret",
type: "string",
description: "Client secret to be used to authenticate calls to the Console",
},
},
handler: async argv => {
const spinner = ora({ text: `Importing Twake data on ${argv.url}` }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
const consoleService = platform.getProvider<ConsoleServiceImpl>("console");
const merge = consoleService.merge(
argv.url,
argv.concurrent,
argv.dry,
argv.console,
argv.link,
argv.client,
argv.secret,
);
const start = Date.now();
let stop: number = start;
const userReports: UserReport[] = [];
const companyReports: CompanyReport[] = [];
const ownerReports: UserReport[] = [];
let companiesWithoutAdmin: Company[] = [];
const process = merge.subscribe({
next: report => {
if (report.type === "company:created") {
const companyReport = report.data as CompanyReport;
companyReports.push(companyReport);
if (companyReport.status === "success") {
spinner.succeed(
`Company created: ${companyReport.company.source.id} ${companyReport.company.source.displayName}`,
);
} else {
spinner.fail(
`Creation error for company ${companyReport.company.source.id} (${companyReport.company.source.displayName}): ${companyReport?.error?.message}`,
);
}
}
if (report.type === "user:created") {
const userReport = report.data as UserReport;
userReports.push(userReport);
if (userReport.status === "success") {
spinner.succeed(
`Company ${userReport.user.source.company.id}: User created ${userReport.user.destination.id}`,
);
} else {
spinner.fail(
`Company ${userReport.user.source.company.id}: User creation error ${userReport.user.destination.id}: ${userReport?.error?.message}`,
);
}
}
if (report.type === "user:updated") {
const userReport = report.data as UserReport;
ownerReports.push(userReport);
if (userReport.status === "success") {
spinner.succeed(
`Company ${userReport.destinationCompanyCode} owner updated to user ${userReport.destinationId}`,
);
} else {
spinner.fail(
`Company ${userReport.destinationCompanyCode}: Owner update error for user ${userReport.destinationId}: ${userReport?.error?.message}`,
);
}
}
if (report.type === "processing:owner") {
spinner.start("Computing company owners, please wait...");
}
if (report.type === "user:updating") {
//spinner.succeed(`Updating owner for company ${report?.company?.sourceId}`);
}
if (report.type === "log") {
report.message && (spinner.text = report.message);
}
if (report.type === "company:withoutadmin") {
report.message && (spinner.text = report.message);
const companies = (report.data || []) as Company[];
if (companies.length) {
companiesWithoutAdmin = companies;
}
}
},
error: () => {
spinner.fail("Fatal error");
},
complete: async () => {
stop = Date.now();
spinner.succeed("Merge is complete");
displayStats();
await tearDown();
},
});
async function tearDown() {
await platform.stop();
process.unsubscribe();
}
function reportAsCSV() {
const users = userReports.map(
user =>
`${user.sourceId},${user.destinationId},${user.destinationCompanyCode},${user.status}`,
);
const userCSV = [...["sourceId,destinationId,destinationCompanyCode,status"], ...users].join(
"\n",
);
const companies = companyReports.map(
company => `${company.sourceId},${company.destinationCode},${company.status}`,
);
const companyCSV = [...["sourceId,destinationCode,status"], ...companies].join("\n");
console.log(userCSV);
console.log(companyCSV);
}
function displayStats() {
const userFailures = userReports.filter(user => user.error);
const companyFailures = companyReports.filter(company => company.error);
const ownerFailures = ownerReports.filter(report => report.error);
console.log("# Import report");
console.log(`Data imported in ${(stop - start) / 1000} seconds`);
console.log("## Company");
console.log("- Companies success:", companyReports.filter(company => !company.error).length);
console.log("- Companies failure:", companyFailures.length);
if (companyFailures.length) {
console.log("### Failures");
companyFailures.forEach(failure =>
console.log(
`- Company ${failure.company.source.id} error: ${failure.company.error?.message}`,
),
);
}
console.log("## User");
console.log("- Users success:", userReports.filter(user => !user.error).length);
console.log("- Users failure:", userFailures.length);
if (userFailures.length) {
console.log("### Failures");
userFailures.forEach(failure =>
console.log(
`- User ${failure.user.source.user.user_id} error: ${failure.user.error?.message}`,
),
);
}
console.log("## Owner");
console.log("- Owners success:", ownerReports.filter(report => !report.error).length);
console.log("- Owners failure:", ownerFailures.length);
if (ownerFailures.length) {
console.log("### Failures");
ownerFailures.forEach(report =>
console.log(`- User ${report.destinationId} error: ${report?.error?.message}`),
);
}
console.log("## Warnings");
console.log("### Companies witout admins");
companiesWithoutAdmin.forEach(company => {
console.log(`- ${company.id} - ${company.displayName}`);
});
if (argv.csv) {
reportAsCSV();
}
}
},
};
export default command;
+14
View File
@@ -0,0 +1,14 @@
import { CommandModule } from "yargs";
const command: CommandModule = {
describe: "Export Twake data",
command: "export <command>",
builder: yargs =>
yargs.commandDir("export_cmds", {
visit: commandModule => commandModule.default,
}),
// eslint-disable-next-line @typescript-eslint/no-empty-function
handler: () => {},
};
export default command;
@@ -0,0 +1,266 @@
import yargs from "yargs";
import twake from "../../../twake";
import { mkdirSync, writeFileSync } from "fs";
import { Pagination } from "../../../core/platform/framework/api/crud-service";
import WorkspaceUser from "../../../services/workspaces/entities/workspace_user";
import { ChannelVisibility } from "../../../services/channels/types";
import { Channel, ChannelMember } from "../../../services/channels/entities";
import { Message } from "../../../services/messages/entities/messages";
import { MessageWithReplies } from "../../../services/messages/types";
import { formatCompany } from "../../../services/user/utils";
import gr from "../../../services/global-resolver";
import { formatUser } from "../../../utils/users";
/**
* Merge command parameters. Check the builder definition below for more details.
*/
type CLIArgs = {
id: string;
};
const services = [
"auth",
"storage",
"counter",
"message-queue",
"user",
"files",
"messages",
"workspaces",
"platform-services",
"console",
"applications",
"search",
"database",
"webserver",
"channels",
"statistics",
];
const command: yargs.CommandModule<unknown, CLIArgs> = {
command: "company",
describe:
"command to export everything inside a company (publicly data only available to a new member)",
builder: {
id: {
default: "",
type: "string",
description: "Company ID",
},
output: {
default: "",
type: "string",
description: "Folder containing the exported data",
},
},
handler: async argv => {
const platform = await twake.run(services);
await gr.doInit(platform);
const company = await gr.services.companies.getCompany({ id: argv.id });
if (!company) {
return "No such company";
}
console.log(`Start export for company ${company.id}`);
const output = (argv.output as string) || `export-${company.id}`;
mkdirSync(output, { recursive: true });
//Company
console.log("- Create company json file");
writeFileSync(`${output}/company.json`, JSON.stringify(formatCompany(company)));
//Workspaces
console.log("- Create workspaces json file");
const workspaces = await gr.services.workspaces.getAllForCompany(company.id);
writeFileSync(`${output}/workspaces.json`, JSON.stringify(workspaces));
//Users
console.log("- Create users json file");
const users = [];
for (const workspace of workspaces) {
const workspace_users = [];
let workspaceUsers: WorkspaceUser[] = [];
let pagination = new Pagination();
do {
const res = await gr.services.workspaces.getUsers(
{ workspaceId: workspace.id },
pagination,
);
workspaceUsers = [...workspaceUsers, ...res.getEntities()];
pagination = res.nextPage as Pagination;
} while (pagination.page_token);
for (const workspaceUser of workspaceUsers) {
const user = await gr.services.users.get({ id: workspaceUser.userId });
if (user) {
users.push(await formatUser(user));
workspace_users.push({ ...workspaceUser, user });
}
}
mkdirSync(`${output}/workspaces/${workspace.id}`, { recursive: true });
writeFileSync(
`${output}/workspaces/${workspace.id}/users.json`,
JSON.stringify(workspace_users),
);
}
writeFileSync(`${output}/users.json`, JSON.stringify(users));
//Applications
console.log("- Create applications json file");
const applications = await gr.services.applications.companyApps.list(
new Pagination(),
{},
{ company: { id: company.id }, user: { id: "", server_request: true } },
);
writeFileSync(`${output}/applications.json`, JSON.stringify(applications));
//Channels
console.log("- Create channels json file");
const directChannels: Channel[] = [];
let allPublicChannels: Channel[] = [];
let pagination = new Pagination();
do {
const page = await gr.services.channels.channels.getDirectChannelsInCompany(
pagination,
company.id,
undefined,
);
for (const channel of page.getEntities()) {
const channelDetail = await gr.services.channels.channels.get(
{
company_id: channel.company_id,
workspace_id: "direct",
id: channel.id,
},
undefined,
);
directChannels.push(channelDetail);
}
pagination = page.nextPage as Pagination;
} while (pagination.page_token);
for (const workspace of workspaces) {
let pagination = new Pagination();
let publicChannels: Channel[] = [];
pagination = new Pagination();
do {
const page = await gr.services.channels.channels.list(
pagination,
{},
{
user: { id: "", server_request: true },
workspace: { workspace_id: workspace.id, company_id: company.id },
},
);
const chans = page.getEntities().filter(c => c.visibility == ChannelVisibility.PUBLIC);
allPublicChannels = [...allPublicChannels, ...chans];
publicChannels = [...publicChannels, ...chans];
pagination = page.nextPage as Pagination;
} while (pagination.page_token);
mkdirSync(`${output}/workspaces/${workspace.id}`, { recursive: true });
writeFileSync(
`${output}/workspaces/${workspace.id}/channels.json`,
JSON.stringify(publicChannels),
);
}
writeFileSync(`${output}/direct_channels.json`, JSON.stringify(directChannels));
//Channels users
console.log("- Create channels users json file");
for (const channel of [...allPublicChannels /*, ...directChannels*/]) {
let members: ChannelMember[] = [];
let pagination = new Pagination();
do {
const page = await gr.services.channels.members.list(
pagination,
{},
{
user: { id: "", server_request: true },
channel: {
company_id: channel.company_id,
workspace_id: channel.workspace_id,
id: channel.id,
},
},
);
members = [...members, ...page.getEntities()] as ChannelMember[];
pagination = page.nextPage as Pagination;
} while (pagination.page_token);
mkdirSync(`${output}/workspaces/${channel.workspace_id}/channels/${channel.id}`, {
recursive: true,
});
writeFileSync(
`${output}/workspaces/${channel.workspace_id}/channels/${channel.id}/members.json`,
JSON.stringify(members),
);
}
//Messages
console.log("- Create messages json file");
//Note: direct channels content is private and not needed for R&D
for (const channel of [...allPublicChannels /*, ...directChannels*/]) {
let threads: MessageWithReplies[] = [];
let messages: Message[] = [];
let pagination = new Pagination();
try {
do {
const page = await gr.services.messages.views.listChannel(
pagination,
{
include_users: false,
replies_per_thread: 10000,
emojis: false,
},
{
user: { id: "", server_request: true },
channel: {
company_id: channel.company_id,
workspace_id: channel.workspace_id,
id: channel.id,
},
},
);
for (const thread of page.getEntities()) {
messages = [...messages, ...thread.last_replies];
}
threads = [
...threads,
...page.getEntities().map(thread => {
thread.last_replies = undefined;
return thread;
}),
];
pagination = page.nextPage as Pagination;
} while (pagination.page_token);
} catch (err) {
console.log(`-- Error on the channel ${channel.id}`);
console.log(err);
}
mkdirSync(`${output}/workspaces/${channel.workspace_id}/channels/${channel.id}`, {
recursive: true,
});
writeFileSync(
`${output}/workspaces/${channel.workspace_id}/channels/${channel.id}/threads.json`,
JSON.stringify(threads),
);
writeFileSync(
`${output}/workspaces/${channel.workspace_id}/channels/${channel.id}/messages.json`,
JSON.stringify(messages),
);
}
await platform.stop();
},
};
export default command;
@@ -0,0 +1,14 @@
import { CommandModule } from "yargs";
const command: CommandModule = {
describe: "Generate Twake data",
command: "generate <command>",
builder: yargs =>
yargs.commandDir("generate_cmds", {
visit: commandModule => commandModule.default,
}),
// eslint-disable-next-line @typescript-eslint/no-empty-function
handler: () => {},
};
export default command;
@@ -0,0 +1,147 @@
import passwordGenerator from "generate-password";
import { from, pipe } from "rxjs";
import { bufferCount, first, map, mergeMap, switchMap, tap } from "rxjs/operators";
import { v1 as uuid } from "uuid";
import yargs from "yargs";
import Company, {
getInstance as getCompanyInstance,
} from "../../../services/user/entities/company";
import CompanyUser from "../../../services/user/entities/company_user";
import twake from "../../../twake";
import User, { getInstance as getUserInstance } from "../../../services/user/entities/user";
import gr from "../../../services/global-resolver";
type CLIArgs = {
company: number;
user: number;
concurrent: number;
};
const services = [
"storage",
"counter",
"applications",
"statistics",
"auth",
"realtime",
"push",
"platform-services",
"user",
"search",
"channels",
"notifications",
"database",
"webserver",
"message-queue",
];
// eslint-disable-next-line @typescript-eslint/ban-types
const command: yargs.CommandModule<{}, CLIArgs> = {
command: "users",
describe: "Generate users and companies",
builder: {
company: {
alias: "c",
default: 3,
type: "number",
description: "Number of companies to generate",
},
user: {
alias: "u",
default: 5,
type: "number",
description: "Number of users to generate in the company",
},
concurrent: {
default: 1,
type: "number",
description: "Number of concurrent creation tasks",
},
},
handler: async argv => {
const concurrentTasks = argv.concurrent;
const nbUsersPerCompany = argv.user;
const nbCompanies = argv.company;
const platform = await twake.run(services);
await gr.doInit(platform);
const companies = getCompanies(nbCompanies);
const createUser = async (userInCompany: {
user: User;
company: Company;
}): Promise<CompanyUser> => {
console.log("Creating user", userInCompany);
const created = await gr.services.users.create(getUserInstance(userInCompany.user));
return (await gr.services.companies.setUserRole(userInCompany.company.id, created.entity.id))
?.entity;
};
const createCompany = (company: Company) => {
console.log("Creating company", company);
return gr.services.companies.createCompany(company);
};
const obsv$ = from(companies).pipe(
// Create companies sequentially
mergeMap(company => createCompany(company), concurrentTasks),
// until we create enough companies
bufferCount(companies.length),
tap(companies => console.log("Created companies", companies.length)),
// for each created company
switchMap(companies => from(companies)),
// generate a set of user for each company
map(company => getUsersForCompany(company, nbUsersPerCompany)),
// Create users sequentially
pipe(
switchMap(userCompanies => from(userCompanies)),
mergeMap(userCompany => createUser(userCompany), concurrentTasks),
// until we reach the number of users in the company
bufferCount(nbUsersPerCompany),
),
// until we reach the number of companies
bufferCount(companies.length),
first(),
);
return obsv$
.toPromise()
.then(() => platform.stop())
.finally(() => console.log("✅ Company users are now created"));
},
};
const getCompanies = (size: number = 10): Company[] => {
return [...Array(size).keys()].map(i =>
getCompanyInstance({
id: uuid(),
name: `twake${i}.app`,
displayName: `My Twake Company #${i}`,
}),
);
};
const getUsersForCompany = (company: Company, size: number = 10) => {
return getUsers(company, size).map(user => ({
user,
company,
}));
};
const getUsers = (company: Company, size: number = 100): User[] => {
return [...Array(size).keys()].map(i => getUser(company, i));
};
const getUser = (company: Company, id: number): User => {
return getUserInstance({
id: uuid(),
first_name: "John",
last_name: `Doe${id}`,
email_canonical: `user${id}@${company.name}`,
password: passwordGenerator.generate({
length: 10,
numbers: true,
}),
});
};
export default command;
@@ -0,0 +1,14 @@
import { CommandModule } from "yargs";
const command: CommandModule = {
describe: "Migrate your php message to node",
command: "migration <command>",
builder: yargs =>
yargs.commandDir("migration_cmds", {
visit: commandModule => commandModule.default,
}),
// eslint-disable-next-line @typescript-eslint/no-empty-function
handler: () => {},
};
export default command;
@@ -0,0 +1,264 @@
import yargs from "yargs";
import twake from "../../../twake";
import ora from "ora";
import { TwakePlatform } from "../../../core/platform/platform";
import { DatabaseServiceAPI } from "../../../core/platform/services/database/api";
import PhpApplication, {
DepreciatedDisplayConfiguration,
TYPE as phpTYPE,
} from "./php-application/php-application-entity";
import { Pagination } from "../../../core/platform/framework/api/crud-service";
import Application, { TYPE } from "../../../services/applications/entities/application";
import gr from "../../../services/global-resolver";
type Options = {
onlyApplication?: string;
replaceExisting?: boolean;
};
class ApplicationMigrator {
database: DatabaseServiceAPI;
constructor(readonly platform: TwakePlatform) {
this.database = this.platform.getProvider<DatabaseServiceAPI>("database");
}
public async run(options: Options = {}): Promise<void> {
const phpRepository = await this.database.getRepository(phpTYPE, PhpApplication);
const repository = await this.database.getRepository(TYPE, Application);
let page: Pagination = { limitStr: "100" };
do {
const applicationListResult = await phpRepository.find({}, { pagination: page }, undefined);
page = applicationListResult.nextPage as Pagination;
for (const application of applicationListResult.getEntities()) {
if (
!(await repository.findOne(
{
id: application.id,
},
{},
undefined,
)) ||
options.replaceExisting
) {
const newApplication = importDepreciatedFields(application);
await repository.save(newApplication, undefined);
}
}
} while (page.page_token);
}
}
const services = [
"storage",
"counter",
"platform-services",
"user",
"search",
"channels",
"database",
"webserver",
"message-queue",
"applications",
"console",
"auth",
"statistics",
"realtime",
];
const command: yargs.CommandModule<unknown, unknown> = {
command: "application",
describe: "command that allow you to migrate php applications to node",
builder: {
onlyApplication: {
default: null,
type: "string",
description: "Migrate only this application ID",
},
replaceExisting: {
default: false,
type: "boolean",
description: "Replace already migrated applications",
},
},
// eslint-disable-next-line @typescript-eslint/no-unused-vars
handler: async argv => {
const spinner = ora({ text: "Migrating php applications - " }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
const migrator = new ApplicationMigrator(platform);
const onlyApplication = argv.onlyApplication as string | null;
const replaceExisting = (argv.replaceExisting || false) as boolean;
await migrator.run({
onlyApplication,
replaceExisting,
});
return spinner.stop();
},
};
export default command;
export const importDepreciatedFields = (application: PhpApplication): Application => {
const newApplication = new Application();
newApplication.id = application.id;
newApplication.company_id = application.group_id;
newApplication.is_default = application.is_default;
if (!newApplication.identity?.name) {
newApplication.identity = {
code:
application.depreciated_simple_name ||
(application.depreciated_name || "").toLocaleLowerCase(),
name: application.depreciated_name,
icon: application.depreciated_icon_url,
description: application.depreciated_description,
website: "http://twake.app/",
categories: [],
compatibility: ["twake"],
};
}
if (newApplication.publication?.published === undefined) {
newApplication.publication = newApplication.publication || {
published: false,
requested: false,
};
newApplication.publication.published = application.depreciated_is_available_to_public;
newApplication.publication.requested =
application.depreciated_public && !application.depreciated_twake_team_validation;
}
if (!newApplication.stats?.version) {
newApplication.stats = newApplication.stats || {
created_at: null,
updated_at: null,
version: null,
};
newApplication.stats.version = 1;
newApplication.stats.created_at = Date.now();
newApplication.stats.updated_at = Date.now();
}
if (!newApplication.api?.private_key) {
newApplication.api = newApplication.api || {
hooks_url: null,
allowed_ips: null,
private_key: null,
};
newApplication.api.hooks_url = application.depreciated_api_events_url;
newApplication.api.allowed_ips = application.depreciated_api_allowed_ip;
newApplication.api.private_key = application.depreciated_api_private_key;
}
if (newApplication.access?.write === undefined) {
newApplication.access = newApplication.access || {
read: null,
write: null,
delete: null,
hooks: null,
};
try {
newApplication.access.write = JSON.parse(application.depreciated_capabilities || "[]") || [];
newApplication.access.delete = JSON.parse(application.depreciated_capabilities || "[]") || [];
} catch (e) {
newApplication.access.write = [];
newApplication.access.delete = [];
}
try {
newApplication.access.read = JSON.parse(application.depreciated_privileges || "[]") || [];
} catch (e) {
newApplication.access.read = [];
}
try {
newApplication.access.hooks = JSON.parse(application.depreciated_hooks || "[]") || [];
} catch (e) {
newApplication.access.hooks = [];
}
}
newApplication.display = importDepreciatedDisplayFields(
newApplication,
JSON.parse(application.depreciated_display_configuration),
);
return newApplication;
};
export const importDepreciatedDisplayFields = (
application: Application,
depreciatedDisplay: DepreciatedDisplayConfiguration,
): Application["display"] => {
let display = application.display;
if (!display?.twake) {
display = display || { twake: {} };
display.twake = display.twake || {};
}
display.twake.tab = depreciatedDisplay?.channel_tab
? { url: depreciatedDisplay?.channel_tab?.iframe } || true
: undefined;
display.twake.standalone = depreciatedDisplay?.app
? { url: depreciatedDisplay?.app?.iframe } || true
: undefined;
display.twake.configuration = [];
if (depreciatedDisplay?.configuration?.can_configure_in_workspace)
display.twake.configuration.push("global");
if (depreciatedDisplay?.configuration?.can_configure_in_channel)
display.twake.configuration.push("channel");
display.twake.direct = depreciatedDisplay?.member_app
? { name: application.identity.name, icon: application.identity.icon } || true
: undefined;
if (depreciatedDisplay?.drive_module) {
display.twake.files = {
editor: undefined,
actions: [],
};
display.twake.files.editor = {
preview_url: depreciatedDisplay?.drive_module?.can_open_files?.preview_url,
edition_url: depreciatedDisplay?.drive_module?.can_open_files?.url,
extensions: [
...(depreciatedDisplay?.drive_module?.can_open_files?.main_ext || []),
...(depreciatedDisplay?.drive_module?.can_open_files?.other_ext || []),
],
empty_files: (depreciatedDisplay?.drive_module?.can_create_files as any) || [],
};
}
if (depreciatedDisplay?.messages_module) {
display.twake.chat = {
input:
depreciatedDisplay?.messages_module?.in_plus ||
depreciatedDisplay?.messages_module?.right_icon
? {
icon: application.identity.icon,
}
: undefined,
commands:
(depreciatedDisplay?.messages_module
?.commands as Application["display"]["twake"]["chat"]["commands"]) || undefined,
actions: depreciatedDisplay?.messages_module?.action
? [
{
name: depreciatedDisplay?.messages_module?.action.description,
id: "default",
},
]
: undefined,
};
}
return display;
};
@@ -0,0 +1,173 @@
import { DatabaseServiceAPI } from "../../../core/platform/services/database/api";
import { TwakePlatform } from "../../../core/platform/platform";
import Repository from "../../../core/platform/services/database/services/orm/repository/repository";
import { ChannelMember, ChannelMemberReadCursors } from "../../../services/channels/entities";
import {
ExecutionContext,
Paginable,
Pagination,
} from "../../../core/platform/framework/api/crud-service";
import gr from "../../../services/global-resolver";
import yargs from "yargs";
import ora from "ora";
import twake from "../../../twake";
class ChannelMemberReadCursorsMigrator {
readSectionRepository: Repository<ChannelMemberReadCursors>;
database: DatabaseServiceAPI;
constructor(readonly _platform: TwakePlatform) {
this.database = this._platform.getProvider<DatabaseServiceAPI>("database");
}
public async run(_options = {}, context?: ExecutionContext): Promise<void> {
const companyPagination: Paginable = new Pagination(null, "100");
const companies = await gr.services.companies.getCompanies(companyPagination);
context.user.server_request = true;
for (const company of companies.getEntities()) {
const { id: CompanyId } = company;
const workspaces = await gr.services.workspaces.getAllForCompany(CompanyId);
const workspaceIds = [...workspaces.map(({ id }) => id), "direct"];
for (const workspaceId of workspaceIds) {
const channels = await gr.services.channels.channels.getAllChannelsInWorkspace(
CompanyId,
workspaceId,
);
for (const channel of channels) {
const { id: channelId } = channel;
const membersPagination: Pagination = new Pagination(null, "100");
const threadsPagination: Pagination = new Pagination(null, "1");
let members: ChannelMember[];
try {
const membersList = await gr.services.channels.members.list(
membersPagination,
{},
{
...context,
channel: {
company_id: CompanyId,
id: channelId,
workspace_id: workspaceId,
},
},
);
members = membersList.getEntities();
} catch (error) {
continue;
}
let lastMessage = await gr.services.messages.views.listChannel(
threadsPagination,
{},
{
...context,
channel: {
company_id: CompanyId,
id: channelId,
workspace_id: workspaceId,
},
user: {
id: null,
},
},
);
const firstMessage = await gr.services.messages.views.listChannel(
{
...threadsPagination,
reversed: true,
},
{},
{
...context,
channel: {
company_id: CompanyId,
id: channelId,
workspace_id: workspaceId,
},
user: {
id: null,
},
},
);
if (!firstMessage.getEntities().length) {
continue;
}
if (!lastMessage.getEntities().length) {
lastMessage = firstMessage;
}
const firstMessageId = firstMessage.getEntities()[0].id;
const lastMessageId = lastMessage.getEntities()[0].id;
if (!firstMessageId) {
continue;
} else {
for (const member of members) {
await gr.services.channels.members.setChannelMemberReadSections(
{
start: firstMessageId,
end: lastMessageId,
},
{
...context,
channel_id: channelId,
workspace_id: workspaceId,
company: {
id: CompanyId,
},
user: {
id: member.user_id,
},
},
);
}
}
}
}
}
}
}
const services = [
"search",
"database",
"webserver",
"auth",
"counter",
"cron",
"message-queue",
"push",
"realtime",
"storage",
"tracker",
"websocket",
"email-pusher",
];
const command: yargs.CommandModule<unknown, unknown> = {
command: "channel-member-read-cursor-repair",
describe: "fixes the channel members read cursors for old messages",
builder: {},
handler: async () => {
const spinner = ora({ text: "Fixing channel members read cursors" }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
const migrator = new ChannelMemberReadCursorsMigrator(platform);
await migrator.run({});
return spinner.stop();
},
};
export default command;
@@ -0,0 +1,103 @@
import yargs from "yargs";
import twake from "../../../twake";
import ora from "ora";
import { TwakePlatform } from "../../../core/platform/platform";
import { DatabaseServiceAPI } from "../../../core/platform/services/database/api";
import { PhpDevice, TYPE as phpTYPE } from "./php-device/php-device-entity";
import { Pagination } from "../../../core/platform/framework/api/crud-service";
import Device, { TYPE } from "../../../services/user/entities/device";
import User, { TYPE as userTYPE } from "../../../services/user/entities/user";
import _ from "lodash";
import gr from "../../../services/global-resolver";
type Options = {
replaceExisting?: boolean;
};
class DeviceMigrator {
database: DatabaseServiceAPI;
constructor(readonly platform: TwakePlatform) {
this.database = this.platform.getProvider<DatabaseServiceAPI>("database");
}
public async run(options: Options = {}): Promise<void> {
const phpRepository = await this.database.getRepository(phpTYPE, PhpDevice);
const userRepository = await this.database.getRepository(userTYPE, User);
const repository = await this.database.getRepository(TYPE, Device);
// Get all companies
let page: Pagination = { limitStr: "100" };
// For each devices
do {
const deviceListResult = await phpRepository.find({}, { pagination: page }, undefined);
page = deviceListResult.nextPage as Pagination;
for (const device of deviceListResult.getEntities()) {
if (
!(await repository.findOne({ id: device.value }, {}, undefined)) ||
options.replaceExisting
) {
if (device.type === "FCM" || device.type === "fcm") {
const newDevice = new Device();
newDevice.id = device.value;
newDevice.type = "FCM";
newDevice.user_id = device.user_id;
newDevice.version = device.version;
await repository.save(newDevice, undefined);
const user = await userRepository.findOne({ id: device.user_id }, {}, undefined);
if (user) {
user.devices = _.uniq([...(user.devices || []), newDevice.id]);
await userRepository.save(user, undefined);
}
}
}
}
} while (page.page_token);
}
}
const services = [
"storage",
"counter",
"platform-services",
"user",
"search",
"channels",
"database",
"webserver",
"message-queue",
"workspaces",
"console",
"auth",
];
const command: yargs.CommandModule<unknown, unknown> = {
command: "device",
describe: "command that allow you to migrate php devices to node",
builder: {
replaceExisting: {
default: false,
type: "boolean",
description: "Replace already migrated devices",
},
},
// eslint-disable-next-line @typescript-eslint/no-unused-vars
handler: async argv => {
const spinner = ora({ text: "Migrating php devices - " }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
const migrator = new DeviceMigrator(platform);
const replaceExisting = (argv.replaceExisting || false) as boolean;
await migrator.run({
replaceExisting,
});
return spinner.stop();
},
};
export default command;
@@ -0,0 +1,40 @@
import ora from "ora";
import globalResolver from "../../../services/global-resolver";
import twake from "../../../twake";
import yargs from "yargs";
import DriveMigrator from "./php-drive-file/drive-migrator-service";
const services = [
"search",
"database",
"webserver",
"auth",
"counter",
"cron",
"message-queue",
"push",
"realtime",
"storage",
"tracker",
"websocket",
"email-pusher",
];
const command: yargs.CommandModule<unknown, unknown> = {
command: "drive",
describe: "migrate php drive items to node",
builder: {},
handler: async _argv => {
console.log("test");
const spinner = ora({ text: "Migrating php drive - " }).start();
const platform = await twake.run(services);
await globalResolver.doInit(platform);
const migrator = new DriveMigrator(platform);
await migrator.run();
return spinner.stop();
},
};
export default command;
@@ -0,0 +1,96 @@
import yargs from "yargs";
import twake from "../../../twake";
import ora from "ora";
import { DatabaseServiceAPI } from "../../../core/platform/services/database/api";
import { TwakePlatform } from "../../../core/platform/platform";
import gr from "../../../services/global-resolver";
type Options = {
from?: string;
onlyCompany?: string;
onlyWorkspace?: string;
onlyChannel?: string;
dryRun?: boolean;
};
class MessageReferenceRepair {
database: DatabaseServiceAPI;
constructor(readonly platform: TwakePlatform) {
this.database = this.platform.getProvider<DatabaseServiceAPI>("database");
}
// eslint-disable-next-line @typescript-eslint/no-unused-vars
public async run(_options: Options = {}): Promise<void> {
//TODO repair messages
}
}
const services = [
"storage",
"counter",
"platform-services",
"user",
"search",
"channels",
"database",
"webserver",
"message-queue",
"messages",
];
const command: yargs.CommandModule<unknown, unknown> = {
command: "message-channel-repair",
describe: "command that allow you to repair messages references in channels",
builder: {
from: {
default: null,
type: "string",
description: "Start migration from this company ID",
},
onlyCompany: {
default: null,
type: "string",
description: "Migrate only this company ID",
},
onlyWorkspace: {
default: null,
type: "string",
description: "Migrate only this workspace ID",
},
onlyChannel: {
default: null,
type: "string",
description: "Migrate only this channel ID",
},
dryRun: {
default: false,
type: "boolean",
description: "Do not save anything and show missing references",
},
},
handler: async argv => {
const spinner = ora({ text: "Fixing messages references - " }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
const migrator = new MessageReferenceRepair(platform);
const from = argv.from as string | null;
const onlyCompany = argv.onlyCompany as string | null;
const onlyWorkspace = argv.onlyWorkspace as string | null;
const onlyChannel = argv.onlyChannel as string | null;
const dryRun = (argv.dryRun || false) as boolean;
await migrator.run({
from,
onlyCompany,
onlyWorkspace,
onlyChannel,
dryRun,
});
return spinner.stop();
},
};
export default command;
@@ -0,0 +1,190 @@
import yargs from "yargs";
import twake from "../../../twake";
import ora from "ora";
import { TwakePlatform } from "../../../core/platform/platform";
import { DatabaseServiceAPI } from "../../../core/platform/services/database/api";
import {
ExecutionContext,
Paginable,
Pagination,
} from "../../../core/platform/framework/api/crud-service";
import gr from "../../../services/global-resolver";
import { getInstance, MessageFileRef } from "../../../services/messages/entities/message-file-refs";
import Repository from "../../../core/platform/services/database/services/orm/repository/repository";
import uuid from "node-uuid";
import { MessageFile } from "../../../services/messages/entities/message-files";
import { ThreadExecutionContext } from "../../../services/messages/types";
type Options = Record<string, unknown>;
class MessageFilesCacheMigrator {
database: DatabaseServiceAPI;
repository: Repository<MessageFileRef>;
messageFileRepository: Repository<MessageFile>;
constructor(readonly platform: TwakePlatform) {
this.database = this.platform.getProvider<DatabaseServiceAPI>("database");
}
// eslint-disable-next-line @typescript-eslint/no-unused-vars
public async run(_options: Options = {}, context?: ExecutionContext): Promise<void> {
this.repository = await gr.database.getRepository<MessageFileRef>(
"message_file_refs",
MessageFileRef,
);
this.messageFileRepository = await gr.database.getRepository<MessageFile>(
"message_files",
MessageFile,
);
let count = 0;
let companyPagination: Paginable = new Pagination(null, "100");
do {
const companyList = await gr.services.companies.getCompanies(companyPagination);
companyPagination = companyList.nextPage;
for (const company of companyList.getEntities()) {
const companyId: string = company.id;
const workspaceList = await gr.services.workspaces.getAllForCompany(companyId);
for (const workspaceId of [...workspaceList.map(w => w.id), "direct"]) {
const channelsList = await gr.services.channels.channels.getAllChannelsInWorkspace(
companyId,
workspaceId,
context,
);
for (const channel of channelsList) {
const channelId = channel.id;
let threadPagination: Paginable = new Pagination(null, "100");
do {
const threadList = await gr.services.messages.views.listChannel(
threadPagination,
{},
{
channel: {
company_id: companyId,
workspace_id: workspaceId,
id: channelId,
},
user: { id: null },
},
);
for (const thread of threadList.getEntities()) {
let messagesPagination: Paginable = new Pagination(null, "100");
do {
const messagesList = await gr.services.messages.messages.list(
messagesPagination,
{},
{
thread: { id: thread.id },
company: { id: companyId },
workspace: { id: workspaceId },
channel: { id: channelId },
user: { id: null },
} as ThreadExecutionContext,
);
messagesPagination = messagesList.nextPage;
messagesPagination.page_token = messagesList.getEntities()[0]?.id;
for (const message of messagesList.getEntities()) {
if (message.files && message.files.length > 0) {
for (const _messageFile of message.files) {
count++;
try {
const messageFile = await this.messageFileRepository.findOne(
{
message_id: message.id,
id: _messageFile.id,
},
{},
undefined,
);
if (messageFile) {
console.log(messageFile.metadata.name);
//Update user uploads
const fileRef = getInstance({
target_type: "user_upload",
target_id: message.user_id,
id: uuid.v1({ msecs: message.created_at }),
created_at: message.created_at,
company_id: companyId,
workspace_id: workspaceId,
channel_id: channelId,
thread_id: message.thread_id,
message_id: message.id,
message_file_id: messageFile.id,
file_id: messageFile.metadata.external_id,
});
await this.repository.save(fileRef, undefined);
//Update messageFileRepository
messageFile.cache = {
company_id: companyId,
workspace_id: workspaceId,
channel_id: channelId,
user_id: message.user_id,
};
messageFile.thread_id = message.thread_id;
await this.messageFileRepository.save(messageFile, undefined);
}
} catch (e) {}
}
}
}
} while (messagesPagination.page_token);
}
threadPagination = threadList.nextPage;
threadPagination.page_token =
threadPagination.page_token &&
threadList.getEntities()?.[threadList.getEntities().length - 1]?.thread_id;
} while (false); // && threadPagination.page_token && threadList.getEntities().length > 50);
}
}
}
console.log("updated messages: ", count);
} while (companyPagination.page_token);
}
}
const services = [
"search",
"database",
"webserver",
"auth",
"counter",
"cron",
"message-queue",
"push",
"realtime",
"storage",
"tracker",
"websocket",
];
const command: yargs.CommandModule<unknown, unknown> = {
command: "message-files-cache",
describe: "command that allow you to fix cache for each message-file refs",
builder: {},
handler: async () => {
const spinner = ora({ text: "Migrating messages - " }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
const migrator = new MessageFilesCacheMigrator(platform);
await migrator.run({});
return spinner.stop();
},
};
export default command;
@@ -0,0 +1,91 @@
import yargs from "yargs";
import twake from "../../../twake";
import ora from "ora";
import MessageMigrator from "./php-message/message-migrator-service";
import gr from "../../../services/global-resolver";
const services = [
"storage",
"counter",
"platform-services",
"user",
"search",
"channels",
"database",
"webserver",
"message-queue",
"messages",
"statistics",
];
const command: yargs.CommandModule<unknown, unknown> = {
command: "message",
describe: "command that allow you to migrate php messages to node",
builder: {
from: {
default: null,
type: "string",
description: "Start migration from this company ID",
},
onlyCompany: {
default: null,
type: "string",
description: "Migrate only this company ID",
},
onlyWorkspace: {
default: null,
type: "string",
description: "Migrate only this workspace ID",
},
onlyChannel: {
default: null,
type: "string",
description: "Migrate only this channel ID",
},
ignoreExisting: {
default: false,
type: "boolean",
description: "Skip existing message ids",
},
backToPhp: {
default: false,
type: "boolean",
description:
"Run the migration back, put node messages into php table (only the one that are not already here)",
},
dryRun: {
default: false,
type: "boolean",
description: "Do not save anything",
},
},
// eslint-disable-next-line @typescript-eslint/no-unused-vars
handler: async argv => {
const spinner = ora({ text: "Migrating php messages - " }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
const migrator = new MessageMigrator(platform);
const from = argv.from as string | null;
const onlyCompany = argv.onlyCompany as string | null;
const onlyWorkspace = argv.onlyWorkspace as string | null;
const onlyChannel = argv.onlyChannel as string | null;
const ignoreExisting = (argv.ignoreExisting || false) as boolean;
const backToPhp = (argv.backToPhp || false) as boolean;
const dryRun = (argv.dryRun || false) as boolean;
await migrator.run({
from,
onlyCompany,
onlyWorkspace,
onlyChannel,
ignoreExisting,
backToPhp,
dryRun,
});
return spinner.stop();
},
};
export default command;
@@ -0,0 +1,125 @@
import yargs from "yargs";
import twake from "../../../twake";
import ora from "ora";
import { DatabaseServiceAPI } from "../../../core/platform/services/database/api";
import { TwakePlatform } from "../../../core/platform/platform";
import gr from "../../../services/global-resolver";
import { MessageFileRef } from "../../../services/messages/entities/message-file-refs";
import { Paginable } from "../../../core/platform/framework/api/crud-service";
import { Pagination } from "../../../core/platform/framework/api/crud-service";
import { fileIsMedia } from "../../../services/files/utils";
import { MessageFile } from "../../../services/messages/entities/message-files";
import _ from "lodash";
type Options = Record<string, unknown>;
class MessageReferenceRepair {
database: DatabaseServiceAPI;
constructor(readonly platform: TwakePlatform) {
this.database = this.platform.getProvider<DatabaseServiceAPI>("database");
}
// eslint-disable-next-line @typescript-eslint/no-unused-vars
public async run(_options: Options = {}): Promise<void> {
const repository = await gr.database.getRepository<MessageFileRef>(
"message_file_refs",
MessageFileRef,
);
const repositoryMessageFile = await gr.database.getRepository<MessageFile>(
"message_files",
MessageFile,
);
let count = 0;
let companyPagination: Paginable = new Pagination(null, "100");
do {
const companyList = await gr.services.companies.getCompanies(companyPagination);
companyPagination = companyList.nextPage;
for (const company of companyList.getEntities()) {
const companyId: string = company.id;
const workspaceList = await gr.services.workspaces.getAllForCompany(companyId);
for (const workspaceId of [...workspaceList.map(w => w.id), "direct"]) {
const channelsList = await gr.services.channels.channels.getAllChannelsInWorkspace(
companyId,
workspaceId,
);
for (const channel of channelsList) {
const channelId = channel.id;
let filePagination: Pagination = new Pagination(null, "100");
do {
const items = await repository.find(
{
target_type: "channel",
target_id: channelId,
company_id: companyId,
},
{ pagination: filePagination },
);
for (const item of items.getEntities()) {
try {
const msgFile = await repositoryMessageFile.findOne({
message_id: item.message_id,
id: item.message_file_id,
});
if (msgFile) {
count++;
const isMedia = fileIsMedia(msgFile);
const ref = _.cloneDeep(item);
ref.target_type = isMedia ? "channel_media" : "channel_file";
await repository.save(ref);
}
} catch (e) {
console.log("Error", e);
}
}
filePagination = new Pagination(items.nextPage.page_token, "100");
} while (filePagination.page_token);
}
}
}
console.log("updated files refs: ", count);
} while (companyPagination.page_token);
}
}
const services = [
"search",
"database",
"webserver",
"auth",
"counter",
"cron",
"message-queue",
"push",
"realtime",
"storage",
"tracker",
"websocket",
];
const command: yargs.CommandModule<unknown, unknown> = {
command: "messages-files-medias-separation",
describe: "command to separate medias and files in messages-files channels refs",
builder: {},
handler: async () => {
const spinner = ora({ text: "Fixing messages references - " }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
const migrator = new MessageReferenceRepair(platform);
await migrator.run({});
return spinner.stop();
},
};
export default command;
@@ -0,0 +1,133 @@
import { Type } from "class-transformer";
import {
Column,
Entity,
} from "../../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "application";
@Entity(TYPE, {
primaryKey: [["group_id"], "app_group_name", "id"],
type: TYPE,
})
export default class PhpApplication {
@Type(() => String)
@Column("id", "timeuuid", { generator: "timeuuid" })
id: string;
@Type(() => String)
@Column("group_id", "timeuuid")
group_id: string;
@Type(() => String)
@Column("app_group_name", "string")
app_group_name = "";
@Column("name", "encoded_string")
depreciated_name: string;
@Column("simple_name", "encoded_string")
depreciated_simple_name: string;
@Column("description", "encoded_string")
depreciated_description: string;
@Column("icon_url", "encoded_string")
depreciated_icon_url: string;
@Column("is_default", "twake_boolean")
is_default: boolean;
@Column("public", "twake_boolean")
depreciated_public: boolean;
@Column("twake_team_validation", "twake_boolean")
depreciated_twake_team_validation: boolean;
@Column("is_available_to_public", "twake_boolean")
depreciated_is_available_to_public: boolean; //Vrai si $public ET $twake_team_validation
@Column("api_events_url", "encoded_string")
depreciated_api_events_url: string;
@Column("api_allowed_ip", "encoded_string")
depreciated_api_allowed_ip: string;
@Column("api_private_key", "encoded_string")
depreciated_api_private_key: string;
@Column("privileges", "encoded_string")
depreciated_privileges = "[]";
@Column("capabilities", "encoded_string")
depreciated_capabilities = "[]";
@Column("hooks", "encoded_string")
depreciated_hooks = "[]";
@Column("display_configuration", "encoded_string")
depreciated_display_configuration = "{}";
}
export type DepreciatedDisplayConfiguration = {
version?: 0; //Legacy
tasks_module?: {
can_connect_to_tasks?: boolean;
};
calendar_module?: {
can_connect_to_calendar?: boolean;
};
drive_module?: {
can_connect_to_directory?: boolean;
can_open_files?: {
url?: string; //Une url à appeler pour éditer le fichier (ouvert dans un onglet)
preview_url?: string; //Une url à appeler pour prévisualiser un fichier (iframe)
main_ext?: string[]; //Extensions principales
other_ext?: string[]; //Extensions secondaires
};
can_create_files?: {
url?: string;
filename?: string;
name?: string;
}[];
};
member_app?: boolean; // Si défini, votre application génèrera un membre
// virtuel dans l'espace de travail avec lequel les
// utilisateurs pourront discuter.
messages_module?: {
in_plus?: {
should_wait_for_popup: boolean;
};
right_icon?: {
icon_url?: string; //If defined replace original icon url of your app
should_wait_for_popup?: boolean;
type?: string; //"file" | "call"
};
action?: {
should_wait_for_popup?: boolean;
description?: string; //Description de l'action, sinon remplacé par le nom de l'app
};
commands?: {
command?: string; // my_app mycommand
description?: string;
}[];
};
channel?: {
can_connect_to_channel?: string;
};
channel_tab?: {
iframe?: string;
};
app?: {
iframe?: string;
plus_btn: {
should_wait_for_popup?: boolean;
};
};
configuration?: {
can_configure_in_workspace?: boolean;
can_configure_in_channel?: boolean;
can_configure_in_calendar?: boolean;
can_configure_in_tasks?: boolean;
};
};
@@ -0,0 +1,26 @@
import {
Column,
Entity,
} from "../../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "device";
@Entity(TYPE, {
primaryKey: [["id"]],
type: TYPE,
})
export class PhpDevice {
@Column("id", "timeuuid")
id: string;
@Column("user_id", "timeuuid")
user_id: string;
@Column("type", "string")
type: string;
@Column("version", "string")
version: string;
@Column("value", "encoded_string")
value: string;
}
@@ -0,0 +1,535 @@
import { logger } from "../../../../core/platform/framework";
import { ExecutionContext, Pagination } from "../../../../core/platform/framework/api/crud-service";
import { TwakePlatform } from "../../../../core/platform/platform";
import Repository from "../../../../core/platform/services/database/services/orm/repository/repository";
import { DriveFile, AccessInformation } from "../../../../services/documents/entities/drive-file";
import {
generateAccessToken,
getDefaultDriveItem,
getDefaultDriveItemVersion,
} from "../../../../services/documents/utils";
import globalResolver from "../../../../services/global-resolver";
import Company from "../../../../services/user/entities/company";
import Workspace from "../../../../services/workspaces/entities/workspace";
import { PhpDriveFile } from "./php-drive-file-entity";
import { PhpDriveFileService } from "./php-drive-service";
import mimes from "../../../../utils/mime";
import WorkspaceUser from "../../../../services/workspaces/entities/workspace_user";
import CompanyUser from "src/services/user/entities/company_user";
interface CompanyExecutionContext extends ExecutionContext {
company: {
id: string;
};
}
interface WorkspaceExecutionContext extends CompanyExecutionContext {
workspace_id: string;
}
class DriveMigrator {
private phpDriveService: PhpDriveFileService;
private nodeRepository: Repository<DriveFile>;
constructor(readonly _platform: TwakePlatform) {
this.phpDriveService = new PhpDriveFileService();
}
/**
* Run
*/
public run = async (): Promise<void> => {
await globalResolver.doInit(this._platform);
await this.phpDriveService.init();
this.nodeRepository = await globalResolver.database.getRepository<DriveFile>(
"drive_files",
DriveFile,
);
let page: Pagination = { limitStr: "100" };
const context: ExecutionContext = {
user: {
id: null,
server_request: true,
},
};
do {
const companyListResult = await globalResolver.services.companies.getCompanies(page);
page = companyListResult.nextPage as Pagination;
for (const company of companyListResult.getEntities()) {
await this.migrateCompany(company, {
...context,
company: { id: company.id },
});
}
} while (page.page_token);
};
/**
* Migrate a company drive files.
*
* @param {Company} company - the company to migrate
*/
private migrateCompany = async (
company: Company,
context: CompanyExecutionContext,
): Promise<void> => {
logger.info(`Migrating company ${company.id}`);
const companyAdminOrOwnerId = await this.getCompanyOwnerOrAdminId(company.id, context);
const workspaceList = await globalResolver.services.workspaces.getAllForCompany(company.id);
for (const workspace of workspaceList) {
const wsContext = {
...context,
workspace_id: workspace.id,
user: { id: companyAdminOrOwnerId, server_request: true },
};
const access = await this.getWorkspaceAccess(workspace, company, wsContext);
await this.migrateWorkspace(workspace, access, wsContext);
}
};
/**
* Migrates a workspace drive files.
*
* @param {Workspace} workspace - the workspace to migrate
*/
private migrateWorkspace = async (
workspace: Workspace,
access: AccessInformation,
context: WorkspaceExecutionContext,
): Promise<void> => {
let page: Pagination = { limitStr: "100" };
console.debug(`Migrating workspace ${workspace.id} of company ${context.company.id}`);
logger.info(`Migrating workspace ${workspace.id} root folder`);
const workspaceFolder = await this.createWorkspaceFolder(workspace, access, context);
// Migrate the root folder.
do {
const phpDriveFiles = await this.phpDriveService.listDirectory(
page,
"",
workspace.id,
context,
);
page = phpDriveFiles.nextPage as Pagination;
for (const phpDriveFile of phpDriveFiles.getEntities()) {
await this.migrateDriveFile(phpDriveFile, workspaceFolder.id, access, context);
}
} while (page.page_token);
logger.info(`Migrating workspace ${workspace.id} trash`);
// Migrate the trash.
page = { limitStr: "100" };
do {
const phpDriveFiles = await this.phpDriveService.listDirectory(page, "trash", workspace.id);
page = phpDriveFiles.nextPage as Pagination;
for (const phpDriveFile of phpDriveFiles.getEntities()) {
await this.migrateDriveFile(phpDriveFile, "trash", access, context);
}
} while (page.page_token);
};
/**
* Migrates a php drive item to a Node drive file
*
* @param {PhpDriveItem} item - the php drive file to migrate.
*/
private migrateDriveFile = async (
item: PhpDriveFile,
parentId: string,
access: AccessInformation,
context: WorkspaceExecutionContext,
): Promise<void> => {
logger.info(`Migrating php drive item ${item.id} - parent: ${parentId ?? "root"}`);
try {
const migrationRecord = await this.phpDriveService.getMigrationRecord(
item.id,
context.company.id,
);
const newDriveItem = getDefaultDriveItem(
{
name: item.name || item.id,
extension: item.extension,
added: item.added.toString(),
content_keywords:
item.content_keywords && item.content_keywords.length
? item.content_keywords.join(",")
: "",
creator: item.creator || context.user.id,
is_directory: item.isdirectory,
is_in_trash: item.isintrash,
description: item.description,
tags: item.tags || [],
parent_id: parentId,
company_id: context.company.id,
access_info: access,
},
context,
);
if (migrationRecord && migrationRecord.company_id === context.company.id) {
console.debug(`${item.id} is already migrated`);
} else {
await this.nodeRepository.save(newDriveItem);
}
if (item.isdirectory) {
const newParentId =
migrationRecord && migrationRecord.company_id === context.company.id
? migrationRecord.new_id
: newDriveItem.id;
let page: Pagination = { limitStr: "100" };
do {
const directoryChildren = await this.phpDriveService.listDirectory(
page,
item.id,
context.workspace_id,
);
page = directoryChildren.nextPage as Pagination;
for (const child of directoryChildren.getEntities()) {
try {
await this.migrateDriveFile(child, newParentId, access, context);
} catch (error) {
logger.error(`Failed to migrate drive item ${child.id}`);
console.error(`Failed to migrate drive item ${child.id}`);
}
}
} while (page.page_token);
} else {
let versionPage: Pagination = { limitStr: "100" };
if (
migrationRecord &&
migrationRecord.item_id === item.id &&
migrationRecord.company_id === context.company.id
) {
logger.info(`item is already migrated - ${item.id} - skipping`);
console.log(`item is already migrated - ${item.id} - skipping`);
return;
}
const mime = mimes[item.extension];
let createdVersions = 0;
do {
const itemVersions = await this.phpDriveService.listItemVersions(
versionPage,
item.id,
context,
);
versionPage = itemVersions.nextPage as Pagination;
for (const version of itemVersions.getEntities()) {
try {
const newVersion = getDefaultDriveItemVersion(
{
creator_id: version.creator_id || context.user.id,
data: version.data,
date_added: +version.date_added,
drive_item_id: newDriveItem.id,
file_size: version.file_size,
filename: version.filename,
key: version.key,
provider: version.provider,
realname: version.realname,
mode: version.mode,
},
context,
);
const file = await this.phpDriveService.migrate(
version.file_id,
item.workspace_id,
version.id,
{
filename: version.filename,
userId: version.creator_id || context.user.id,
totalSize: version.file_size,
waitForThumbnail: true,
chunkNumber: 1,
totalChunks: 1,
type: mime,
},
context,
);
if (!file) {
throw Error("cannot download file version");
}
newVersion.file_metadata = {
external_id: file.id,
mime: file.metadata.mime,
name: file.metadata.name || version.filename,
size: file.upload_data.size || version.file_size,
};
await globalResolver.services.documents.documents.createVersion(
newDriveItem.id,
newVersion,
context,
);
createdVersions++;
} catch (error) {
logger.error(`Failed to migrate version ${version.id} for drive item ${item.id}`);
console.error(`Failed to migrate version ${version.id} for drive item ${item.id}`);
}
}
} while (versionPage.page_token);
if (createdVersions === 0) {
await this.nodeRepository.remove(newDriveItem);
return;
}
}
if (!migrationRecord) {
await this.phpDriveService.markAsMigrated(item.id, newDriveItem.id, context.company.id);
}
} catch (error) {
logger.error(
`Failed to migrate Drive item ${item.id} / workspace ${item.workspace_id} / company_id: ${context.company.id}`,
error,
);
console.error(`Failed to migrate Drive item ${item.id}`, error);
}
};
/**
* Fetches the first found company owner or admin identifier.
*
* @param {string} companyId - the companyId
* @param {ExecutionContext} context - the execution context
* @returns {Promise<string>}
*/
private getCompanyOwnerOrAdminId = async (
companyId: string,
context: ExecutionContext,
): Promise<string> => {
let pagination: Pagination = { limitStr: "100" };
let companyOwnerOrAdminId = null;
do {
const companyUsers = await globalResolver.services.companies.companyUserRepository.find(
{ group_id: companyId },
{ pagination },
context,
);
pagination = companyUsers.nextPage as Pagination;
const companyAdminOrOwner = companyUsers
.getEntities()
.find(({ role }) => ["admin", "owner"].includes(role));
if (companyAdminOrOwner) {
companyOwnerOrAdminId = companyAdminOrOwner.id;
}
} while (pagination && !companyOwnerOrAdminId);
return companyOwnerOrAdminId;
};
/**
* Compute the Access Information for the workspace folder to be created.
*
* @param {Workspace} workspace - the target workspace
* @param {Company} company - the target company
* @param {WorkspaceExecutionContext} context - the execution context
* @returns {Promise<AccessInformation>}
*/
private getWorkspaceAccess = async (
workspace: Workspace,
company: Company,
context: WorkspaceExecutionContext,
): Promise<AccessInformation> => {
const companyUsersCount = await globalResolver.services.companies.getUsersCount(company.id);
const workspaceUsersCount = await globalResolver.services.workspaces.getUsersCount(
workspace.id,
);
if (companyUsersCount === workspaceUsersCount) {
return {
entities: [
{
id: "parent",
type: "folder",
level: "manage",
},
{
id: company.id,
type: "company",
level: "none",
},
{
id: context.user?.id,
type: "user",
level: "manage",
},
],
public: {
level: "none",
token: generateAccessToken(),
},
};
}
let workspaceUsers: WorkspaceUser[] = [];
let wsUsersPagination: Pagination = { limitStr: "100" };
do {
const wsUsersQuery = await globalResolver.services.workspaces.getUsers(
{ workspaceId: workspace.id },
wsUsersPagination,
context,
);
wsUsersPagination = wsUsersQuery.nextPage as Pagination;
workspaceUsers = [...workspaceUsers, ...wsUsersQuery.getEntities()];
} while (wsUsersPagination.page_token);
if (companyUsersCount < 30 || workspaceUsersCount < 30) {
return {
entities: [
{
id: "parent",
type: "folder",
level: "none",
},
{
id: company.id,
type: "company",
level: "none",
},
{
id: context.user?.id,
type: "user",
level: "manage",
},
...workspaceUsers.reduce((acc, curr) => {
acc = [
...acc,
{
id: curr.userId,
type: "user",
level: "manage",
},
];
return acc;
}, []),
],
public: {
level: "none",
token: generateAccessToken(),
},
};
}
let companyUsers: CompanyUser[] = [];
let companyUsersPaginations: Pagination = { limitStr: "100" };
do {
const companyUsersQuery = await globalResolver.services.companies.getUsers(
{ group_id: company.id },
companyUsersPaginations,
{},
context,
);
companyUsersPaginations = companyUsersQuery.nextPage as Pagination;
companyUsers = [...companyUsers, ...companyUsersQuery.getEntities()];
} while (companyUsersPaginations.page_token);
return {
entities: [
{
id: "parent",
type: "folder",
level: "none",
},
{
id: company.id,
type: "company",
level: "manage",
},
{
id: context.user?.id,
type: "user",
level: "manage",
},
...companyUsers.reduce((acc, curr) => {
if (workspaceUsers.find(({ userId }) => curr.user_id === userId)) {
return acc;
}
acc = [
...acc,
{
id: curr.user_id,
type: "user",
level: "none",
},
];
return acc;
}, []),
],
public: {
level: "none",
token: generateAccessToken(),
},
};
};
/**
* Creates a folder for the workspace to migrate.
*
* @param {Workspace} workspace - the workspace to migrate.
* @param {AccessInformation} access - the access information.
* @param {WorkspaceExecutionContext} context - the execution context
* @returns {Promise<DriveFile>}
*/
private createWorkspaceFolder = async (
workspace: Workspace,
access: AccessInformation,
context: WorkspaceExecutionContext,
): Promise<DriveFile> => {
const workspaceFolder = getDefaultDriveItem(
{
name: workspace.name || workspace.id,
extension: "",
content_keywords: "",
creator: context.user.id,
is_directory: true,
is_in_trash: false,
description: "",
tags: [],
parent_id: "root",
company_id: context.company.id,
access_info: access,
},
context,
);
await this.nodeRepository.save(workspaceFolder);
await this.phpDriveService.markAsMigrated(workspace.id, workspaceFolder.id, context.company.id);
return workspaceFolder;
};
}
export default DriveMigrator;
@@ -0,0 +1,64 @@
import { Type } from "class-transformer";
import {
Column,
Entity,
} from "../../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "drive_file";
@Entity(TYPE, {
primaryKey: [["workspace_id"], "parent_id", "id"],
type: TYPE,
})
export class PhpDriveFile {
@Type(() => String)
@Column("workspace_id", "timeuuid")
workspace_id: string;
@Type(() => String)
@Column("parent_id", "string")
parent_id: string | "";
@Type(() => String)
@Column("id", "timeuuid", { generator: "timeuuid" })
id: string;
@Column("isintrash", "boolean")
isintrash: boolean;
@Column("added", "string")
added: string;
@Column("attachements", "encoded_json")
attachements: unknown;
@Column("content_keywords", "encoded_json")
content_keywords: string[] | null;
@Column("creator", "string")
creator: string;
@Column("description", "string")
description: string;
@Column("extension", "string")
extension: string;
@Column("hidden_data", "encoded_json")
hidden_data: unknown;
@Column("isdirectory", "boolean")
isdirectory: boolean;
@Column("last_modified", "string")
last_modified: string;
@Column("name", "encoded_string")
name: string;
@Column("size", "number")
size: number;
@Column("tags", "encoded_json")
tags: string[] | null;
}
@@ -0,0 +1,58 @@
import { Type } from "class-transformer";
import {
Column,
Entity,
} from "../../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "drive_file_version";
@Entity(TYPE, {
primaryKey: [["file_id"], "id"],
type: TYPE,
})
export class PhpDriveFileVersion {
@Type(() => String)
@Column("id", "timeuuid", { generator: "timeuuid" })
id: string;
@Type(() => String)
@Column("file_id", "string")
file_id: string;
@Type(() => String)
@Column("creator_id", "string")
creator_id: string;
@Type(() => String)
@Column("realname", "encoded_string")
realname: string;
@Type(() => String)
@Column("key", "string")
key: string;
@Type(() => String)
@Column("mode", "string")
mode: string;
@Type(() => String)
@Column("file_size", "number")
file_size: number;
@Type(() => String)
@Column("date_added", "string")
date_added: string;
@Type(() => String)
@Column("filename", "encoded_string")
filename: string;
@Type(() => String)
@Column("provider", "string")
provider: string;
@Column("data", "encoded_json")
data: unknown;
}
export type PhpDriveFileVersionPrimaryKey = Pick<PhpDriveFileVersion, "file_id" | "id">;
@@ -0,0 +1,25 @@
import { Type } from "class-transformer";
import {
Column,
Entity,
} from "../../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "php_drive_migration_record";
@Entity(TYPE, {
primaryKey: [["company_id"], "item_id"],
type: TYPE,
})
export class phpDriveMigrationRecord {
@Type(() => String)
@Column("item_id", "timeuuid")
item_id: string;
@Type(() => String)
@Column("company_id", "uuid")
company_id: string;
@Type(() => String)
@Column("new_id", "string")
new_id: string;
}
@@ -0,0 +1,189 @@
import {
ExecutionContext,
ListResult,
Pagination,
} from "../../../../core/platform/framework/api/crud-service";
import Repository from "../../../../core/platform/services/database/services/orm/repository/repository";
import globalResolver from "../../../../services/global-resolver";
import { PhpDriveFile, TYPE as DRIVE_FILE_TABLE } from "./php-drive-file-entity";
import { Initializable, logger, TwakeServiceProvider } from "../../../../core/platform/framework";
import {
PhpDriveFileVersion,
TYPE as DRIVE_FILE_VERSION_TABLE,
} from "./php-drive-file-version-entity";
import axios from "axios";
import { Multipart } from "fastify-multipart";
import { CompanyExecutionContext } from "../../../../services/files/web/types";
import { File } from "../../../../services/files/entities/file";
import { UploadOptions } from "../../../../services/files/types";
import {
phpDriveMigrationRecord,
TYPE as MIGRATION_RECORD_TABLE,
} from "./php-drive-migration-record-entity";
export interface MigrateOptions extends UploadOptions {
userId: string;
}
export interface PhpDriveServiceAPI extends TwakeServiceProvider, Initializable {}
export class PhpDriveFileService implements PhpDriveServiceAPI {
version: "1";
public repository: Repository<PhpDriveFile>;
public versionRepository: Repository<PhpDriveFileVersion>;
public migrationRepository: Repository<phpDriveMigrationRecord>;
/**
* Init the service.
*
* @returns {PhpDriveFileService}
*/
async init(): Promise<this> {
this.repository = await globalResolver.database.getRepository<PhpDriveFile>(
DRIVE_FILE_TABLE,
PhpDriveFile,
);
this.versionRepository = await globalResolver.database.getRepository<PhpDriveFileVersion>(
DRIVE_FILE_VERSION_TABLE,
PhpDriveFileVersion,
);
this.migrationRepository = await globalResolver.database.getRepository<phpDriveMigrationRecord>(
MIGRATION_RECORD_TABLE,
phpDriveMigrationRecord,
);
return this;
}
/**
* Lists the drive item directory children.
*
* @param {Pagination} pagination - the page.
* @param {string} directory - the drive item / directory id to search within.
* @param {string} workspaceId - the workspace id
* @param {ExecutionContext} context - the execution context.
* @returns {Promise<ListResult<PhpDriveFile>>} - the drive item children.
*/
listDirectory = async (
pagination: Pagination,
directory: string | "" | "trash",
workspaceId: string,
context?: ExecutionContext,
): Promise<ListResult<PhpDriveFile>> =>
await this.repository.find(
{
workspace_id: workspaceId,
parent_id: directory,
},
{ pagination },
context,
);
/**
* Lists the versions of a drive item.
*
* @param {Pagination} pagination - the page.
* @param {string} itemId - the drive item id.
* @param {ExecutionContext} context - the execution context.
* @returns {Promise<ListResult<PhpDriveFileVersion>>} - the list of the item versions.
*/
listItemVersions = async (
pagination: Pagination,
itemId: string,
context?: ExecutionContext,
): Promise<ListResult<PhpDriveFileVersion>> =>
await this.versionRepository.find(
{
file_id: itemId,
},
{ pagination },
context,
);
/**
* Downloads a file version from the old drive and uploads it to the new Drive.
*
* @param {string} fileId - the old file id
* @param {string} workspaceId - the workspace id
* @param {string} versionId - the version id
* @param {MigrateOptions} options - the file upload / migration options.
* @param {CompanyExecutionContext} context - the company execution context.
* @param {string} public_access_key - the file public access key.
* @returns {Promise<File>} - the uploaded file information.
*/
migrate = async (
fileId: string,
workspaceId: string,
versionId: string,
options: MigrateOptions,
context: CompanyExecutionContext,
public_access_key?: string,
): Promise<File> => {
try {
const url = `https://web.twake.app/ajax/drive/download?workspace_id=${workspaceId}&element_id=${fileId}&version_id=${versionId}&download=1${
public_access_key ? `&public_access_key=${public_access_key}` : ""
}`;
const response = await axios.get(url, {
responseType: "stream",
});
if (!response.data) {
throw Error("invalid download response");
}
const file = {
file: response.data,
};
return await globalResolver.services.files.save(null, file as Multipart, options, {
...context,
user: {
id: options.userId,
},
});
} catch (error) {
logger.error(`Failed to migrate file ${fileId} on workspace ${workspaceId}`, error);
throw Error(error);
}
};
/**
* Saves a drive item.
*
* @param {PhpDriveFile} item - the php drive item.
* @returns {Promise<void>}
*/
save = async (item: PhpDriveFile): Promise<void> => await this.repository.save(item);
/**
* Marks a drive item as migrated.
*
* @param {string} itemId - the drive item.
* @param {string} newId - the new drive item id.
* @param {string} companyId - the company id.
*/
markAsMigrated = async (itemId: string, newId: string, companyId: string): Promise<void> => {
const migrationRecord = new phpDriveMigrationRecord();
migrationRecord.item_id = itemId;
migrationRecord.new_id = newId;
migrationRecord.company_id = companyId;
await this.migrationRepository.save(migrationRecord);
};
/**
* Fetches the drive item migration record.
*
* @param {string} itemId - the drive item id.
* @param {string} companyId - the company id.
* @returns {Promise<boolean>}
*/
getMigrationRecord = async (
itemId: string,
companyId: string,
): Promise<phpDriveMigrationRecord> =>
await this.migrationRepository.findOne({ item_id: itemId, company_id: companyId });
}
@@ -0,0 +1,560 @@
import { PhpMessagesService } from "./php-message-service";
import { convertUuidV4ToV1 } from "./utils";
import Company from "../../../../services/user/entities/company";
import { Pagination } from "../../../../core/platform/framework/api/crud-service";
import { PhpMessage } from "./php-message-entity";
import { TwakePlatform } from "../../../../core/platform/platform";
import {
Message,
MessageEdited,
MessageOverride,
MessagePinnedInfo,
MessageReaction,
} from "../../../../services/messages/entities/messages";
import { MessageChannelRef } from "../../../../services/messages/entities/message-channel-refs";
import { ParticipantObject, Thread } from "../../../../services/messages/entities/threads";
import { Block } from "../../../../services/messages/blocks-types";
import { WorkspaceExecutionContext } from "../../../../services/workspaces/types";
import gr from "../../../../services/global-resolver";
type MigratedChannel = {
id: string;
workspace_id: string;
company_id: string;
owner?: string;
};
type Options = {
from?: string;
onlyCompany?: string;
onlyWorkspace?: string;
onlyChannel?: string;
ignoreExisting?: boolean;
backToPhp?: boolean;
dryRun?: boolean;
};
class MessageMigrator {
private phpMessageService: PhpMessagesService;
private migratedMessages = 0;
private options: Options = {};
constructor(readonly _platform: TwakePlatform) {
this.phpMessageService = new PhpMessagesService();
}
public async run(options: Options = {}): Promise<void> {
await gr.doInit(this._platform);
this.options = options;
await this.phpMessageService.init();
if (this.options.onlyCompany) {
const company = await gr.services.companies.getCompany({ id: options.onlyCompany });
await this.migrateCompanyMessages(company);
} else {
let waitForCompany = false;
if (this.options.from) {
waitForCompany = true;
}
// Get all companies
let page: Pagination = { limitStr: "100" };
// For each companies find workspaces
do {
const companyListResult = await gr.services.companies.getCompanies(page);
page = companyListResult.nextPage as Pagination;
for (const company of companyListResult.getEntities()) {
if (waitForCompany && this.options.from == `${company.id}`) {
waitForCompany = false;
}
if (!waitForCompany) {
await this.migrateCompanyMessages(company);
}
}
} while (page.page_token);
}
console.log(
`Php Messages successfully migrated to node: (${this.migratedMessages} messages) !`,
);
}
private async migrateCompanyMessages(company: Company) {
console.log(`Start migration for ${company.id}...`);
if (!this.options.onlyWorkspace || this.options.onlyWorkspace === "direct") {
await this.migrateCompanyDirectMessages(company);
console.log(
`${company.id} - (1/2) migrated direct messages (total: ${this.migratedMessages} messages) ✅`,
);
}
if (!this.options.onlyWorkspace || this.options.onlyWorkspace !== "direct") {
await this.migrateCompanyChannelsMessages(company);
console.log(`${company.id} - (2/2) completed (total: ${this.migratedMessages} messages) ✅`);
}
}
/**
* Set all direct messages in company and set them to channelPhpMessages
*/
private async migrateCompanyDirectMessages(company: Company) {
await gr.doInit(this._platform);
let pageDirectChannels: Pagination = { limitStr: "100" };
// For each directChannels find messages
do {
const directChannelsInCompanyResult = await gr.services.channels.channels.list(
pageDirectChannels,
{},
{
workspace: {
workspace_id: "direct",
company_id: company.id,
},
user: { id: null, server_request: true },
},
);
pageDirectChannels = directChannelsInCompanyResult.nextPage as Pagination;
for (const directChannel of directChannelsInCompanyResult.getEntities()) {
await this.migrateChannelsMessages(company, {
id: directChannel.id,
workspace_id: "direct",
company_id: directChannel.company_id,
});
}
} while (pageDirectChannels.page_token);
}
/**
* Set all messages in company and set them to channelPhpMessages
*/
private async migrateCompanyChannelsMessages(company: Company) {
await gr.doInit(this._platform);
// Get all workspaces in company
const workspacesInCompany = (
await gr.services.workspaces.list({ limitStr: "" }, {}, {
user: {
id: null,
server_request: true,
},
company_id: company.id,
} as WorkspaceExecutionContext)
).getEntities();
// For each workspaces find channels
for (const workspace of workspacesInCompany) {
if (this.options.onlyWorkspace && `${workspace.id}` !== this.options.onlyWorkspace) {
continue;
}
// Get all channels in workspace
let pageChannels: Pagination = { limitStr: "1" };
do {
const channelsInWorkspace = await gr.services.channels.channels.list(
pageChannels,
{},
{
workspace: {
workspace_id: workspace.id,
company_id: workspace.company_id,
},
user: { id: null, server_request: true },
},
);
pageChannels = channelsInWorkspace.nextPage as Pagination;
// For each channels find messages
for (const channel of channelsInWorkspace.getEntities()) {
await this.migrateChannelsMessages(company, {
id: channel.id,
workspace_id: channel.workspace_id,
company_id: channel.company_id,
owner: channel.owner,
});
}
} while (pageChannels.page_token);
}
}
//Params: company, channel
private async migrateChannelsMessages(company: Company, channel: MigratedChannel) {
if (this.options.onlyChannel && `${channel.id}` !== this.options.onlyChannel) {
return;
}
if (this.options.backToPhp) {
await this.migrateChannelsMessagesBackToPhp(company, channel);
return;
}
//This function will migrate all messages in a channel
let pagePhpMessages: Pagination = { limitStr: "100" };
do {
const messages = await this.phpMessageService.list(
pagePhpMessages,
{},
{
channel_id: convertUuidV4ToV1(channel.id),
user: { id: null, server_request: true },
},
);
for (const message of messages.getEntities()) {
try {
await this.migrateMessage(company, channel, message);
} catch (err) {
console.log(err);
}
}
pagePhpMessages = messages.nextPage as Pagination;
} while (pagePhpMessages.page_token);
}
private migratedThreads: string[] = [];
/**
* Migrate php message to node
* @param company
* @param channel
* @param message
*/
private async migrateMessage(company: Company, channel: MigratedChannel, message: PhpMessage) {
if (!message.id) {
return;
}
await gr.doInit(this._platform);
//Create thread first if not exists
const threadId = message.parent_message_id || message.id;
if (this.options.ignoreExisting) {
const msg = await gr.services.messages.messages.get({
thread_id: threadId,
id: message.id,
});
if (msg) {
return;
}
}
const threadDoesNotExists = !this.migratedThreads.includes(threadId);
if (threadDoesNotExists) {
await this.migratePhpMessageToNodeThread(message, channel, company);
this.migratedThreads.push(threadId);
}
//Migrate message itself
await this.migratePhpMessageToNodeMessage(threadId, message, company);
this.migratedMessages++;
if (this.migratedMessages % 100 == 0) {
console.log(`${company.id} - ... (total: ${this.migratedMessages} messages)`);
}
//Force delay between channels
await new Promise(r => {
setTimeout(r, 40);
});
}
private async migrateChannelsMessagesBackToPhp(company: Company, channel: MigratedChannel) {
await gr.doInit(this._platform);
const channelRefRepository = await gr.database.getRepository(
"message_channel_refs",
MessageChannelRef,
);
const messageRepository = await gr.database.getRepository("messages", Message);
//This function will migrate all messages in a channel
let pageMessages: Pagination = { limitStr: "100" };
do {
const messages = await channelRefRepository.find(
{
company_id: company.id,
workspace_id: channel.workspace_id,
channel_id: channel.id,
},
{ pagination: pageMessages },
undefined,
);
for (const messageRef of messages.getEntities()) {
const messages = await messageRepository.find(
{
thread_id: messageRef.thread_id,
},
{},
undefined,
);
for (const message of messages.getEntities()) {
const uuidv1_channel_id =
channel.id.substring(0, 14) + "1" + channel.id.substring(14 + 1);
let phpMessage = await this.phpMessageService.get({
id: message.id,
});
if (!phpMessage) {
phpMessage = await this.phpMessageService.get({
channel_id: uuidv1_channel_id,
parent_message_id: message.thread_id,
id: message.id,
});
}
if (!phpMessage && message.subtype !== "deleted") {
//This message doesn't exists in php, move it to php
const newPhpMessage = new PhpMessage();
newPhpMessage.id = message.id;
newPhpMessage.channel_id = uuidv1_channel_id;
newPhpMessage.parent_message_id = message.thread_id;
newPhpMessage.application_id = message.application_id || null;
newPhpMessage.modification_date = Math.floor(
message.edited?.edited_at || message.created_at,
);
newPhpMessage.creation_date = Math.floor(message.created_at);
newPhpMessage.sender = message.user_id;
newPhpMessage.pinned = !!message.pinned_info?.pinned_at;
newPhpMessage.edited = !!message.edited?.edited_at;
newPhpMessage.message_type =
message.subtype === "application" ? 1 : message.subtype === "system" ? 2 : 0;
newPhpMessage.hidden_data = message.context;
newPhpMessage.reactions = "{}";
newPhpMessage.responses_count = 0;
let prepared: any[] = [];
(message.blocks || []).map(block => {
if (block.type === "twacode") {
prepared = block["elements"];
}
if (block.type === "section" && prepared.length > 0) {
prepared = [
{
type: "twacode",
content: block?.text?.text || "",
},
];
}
});
newPhpMessage.content = {
fallback_string: message.text,
original_str: message.text,
files: (message.files || []).map(f => {
return {
content: f,
mode: "mini",
type: "file",
};
}),
prepared: prepared,
};
if (!this.options.dryRun) {
await this.phpMessageService.repository.save(newPhpMessage, undefined);
}
this.migratedMessages++;
if (this.migratedMessages % 100 == 0) {
console.log(`${company.id} - ... (total: ${this.migratedMessages} messages)`);
}
}
}
}
pageMessages = messages.nextPage as Pagination;
} while (pageMessages.page_token);
}
/**
* Migrate php message to node thread
* @param message PhpMessage
* @param channel MigratedChannel
*/
private async migratePhpMessageToNodeThread(
message: PhpMessage,
channel: MigratedChannel,
company: Company,
) {
const thread = new Thread();
// Set nodeThread values
thread.id = message.parent_message_id || message.id;
thread.created_at = message.creation_date;
thread.last_activity = message.modification_date;
thread.answers = 0;
thread.participants = [
{
type: "channel",
id: channel.id,
company_id: channel.company_id,
workspace_id: channel.workspace_id,
created_at: message.creation_date,
created_by: message.sender,
} as ParticipantObject,
];
if (this.options.dryRun) {
return;
}
// Create nodeThread
return await gr.services.messages.threads.save(
thread,
{},
{ user: { id: null, server_request: true }, company },
);
}
/**
* Set message string type
* @param integer 0 = null | 1 = system | 2 = application
*/
private setMessageType(integer: number): { type: Message["type"]; subtype: Message["subtype"] } {
switch (integer) {
case 1:
return {
type: "message",
subtype: "application",
};
case 2:
return {
type: "message",
subtype: "system",
};
case 0:
default:
return {
type: "message",
subtype: null,
};
}
}
/**
* Set blocks array
* @param content { [key: string]: any }
*/
// eslint-disable-next-line @typescript-eslint/no-explicit-any
private setBlocks(content: { [key: string]: any }): Block[] {
const blocks: Block[] = [];
if (!content) return blocks;
const new_block_format: Block = {
type: "twacode",
elements: content.formatted || content.prepared || content || [],
};
blocks.push(new_block_format);
return blocks;
}
/**
* Set edited message Object
* @param modification_date timestamp
*/
private setMessageEditedObject(message: PhpMessage): MessageEdited {
return message.edited ? { edited_at: message.modification_date } : null;
}
/**
* Set pinned message Object
* @param message PhpMessage
*/
private setMessagePinnedObject(message: PhpMessage): MessagePinnedInfo {
if (message.pinned) {
return { pinned_at: 0, pinned_by: message.sender };
}
return null;
}
/**
* Set reactions message object
* @param reactions JSON
*/
private setMessageReactionsObject(reactions: string): MessageReaction[] {
const parsed_reactions = JSON.parse(reactions);
if (!parsed_reactions) return [];
const new_reactions_array: MessageReaction[] = [];
for (const reaction_name in parsed_reactions) {
const new_reaction_object: MessageReaction = {
name: reaction_name,
users: parsed_reactions[reaction_name].users,
count: parsed_reactions[reaction_name].count,
};
new_reactions_array.push(new_reaction_object);
}
return new_reactions_array;
}
/**
* Set override message Object
* @param message PhpMessage
*/
private setMessageOverrideObject(_message: PhpMessage): MessageOverride {
return null;
}
/**
* Migrate php message to node message
* @param message PhpMessage
*/
private async migratePhpMessageToNodeMessage(
threadId: string,
message: PhpMessage,
company: Company,
) {
const nodeMessage = new Message();
// Set nodeMessage values
nodeMessage.id = message.id;
nodeMessage.thread_id = threadId;
nodeMessage.type = this.setMessageType(message.message_type).type;
nodeMessage.subtype = this.setMessageType(message.message_type).subtype;
nodeMessage.created_at = message.creation_date;
nodeMessage.user_id = message.sender;
nodeMessage.application_id = message.application_id;
nodeMessage.text = message.content?.original_str || "";
nodeMessage.blocks = this.setBlocks(message.content);
nodeMessage.context = message.hidden_data || {};
nodeMessage.edited = this.setMessageEditedObject(message);
nodeMessage.pinned_info = this.setMessagePinnedObject(message);
nodeMessage.reactions = this.setMessageReactionsObject(message.reactions);
nodeMessage.override = this.setMessageOverrideObject(message);
nodeMessage.context._front_id = message.id;
if (this.options.dryRun) {
return;
}
// Create nodeMessage then add it to thread
return await gr.services.messages.messages.save(
nodeMessage,
{
enforceViewPropagation: true,
},
{
user: { id: null, server_request: true },
thread: { id: threadId },
company: { id: company.id },
},
);
}
}
export default MessageMigrator;
@@ -0,0 +1,67 @@
import { Type } from "class-transformer";
import { merge } from "lodash";
import {
Column,
Entity,
} from "../../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "message";
@Entity(TYPE, {
primaryKey: [["channel_id"], "parent_message_id", "id"],
type: TYPE,
})
export class PhpMessage {
@Type(() => String)
@Column("id", "timeuuid", { generator: "timeuuid" })
id: string;
@Type(() => String)
@Column("channel_id", "timeuuid")
channel_id: string;
@Type(() => String)
@Column("parent_message_id", "encoded_string")
parent_message_id: string;
@Type(() => String)
@Column("application_id", "encoded_string")
application_id: string | null;
@Column("modification_date", "twake_datetime")
modification_date: number | null;
@Column("creation_date", "twake_datetime")
creation_date: number;
@Column("message_type", "number")
message_type: number;
@Column("sender_id", "timeuuid")
sender: string;
@Column("content", "encoded_json")
// eslint-disable-next-line @typescript-eslint/no-explicit-any
content: { [key: string]: any } | null;
@Column("hidden_data", "encoded_json")
// eslint-disable-next-line @typescript-eslint/no-explicit-any
hidden_data: { [key: string]: any } | null;
@Column("reactions", "encoded_string")
reactions: string | null;
@Column("pinned", "twake_boolean")
pinned: boolean;
@Column("edited", "twake_boolean")
edited: boolean;
@Column("responses_count", "twake_int")
responses_count: number | null;
}
export type PhpMessagePrimaryKey = Pick<PhpMessage, "parent_message_id" | "channel_id" | "id">;
export function getInstance(message: PhpMessage): PhpMessage {
return merge(new PhpMessage(), message);
}
@@ -0,0 +1,59 @@
import {
DeleteResult,
ExecutionContext,
ListResult,
Pagination,
} from "../../../../core/platform/framework/api/crud-service";
import Repository, {
FindFilter,
} from "../../../../core/platform/services/database/services/orm/repository/repository";
import { PhpMessagesServiceAPI } from "./types";
import { PhpMessage, PhpMessagePrimaryKey } from "./php-message-entity";
import gr from "../../../../services/global-resolver";
export interface PhpMessageExecutionContext extends ExecutionContext {
channel_id: string;
parent_message_id?: string;
id?: string;
}
export class PhpMessagesService implements PhpMessagesServiceAPI {
version: "1";
public repository: Repository<PhpMessage>;
async init(): Promise<this> {
this.repository = await gr.database.getRepository<PhpMessage>("message", PhpMessage);
return this;
}
get(pk: { parent_message_id?: string; channel_id?: string; id: string }): Promise<PhpMessage> {
if (pk.channel_id) {
pk.channel_id = `${pk.channel_id}`;
pk.channel_id.substring(0, 14) + "1" + pk.channel_id.substring(14 + 1);
}
return this.repository.findOne(pk, {}, undefined);
}
// eslint-disable-next-line @typescript-eslint/no-unused-vars
async delete(pk: PhpMessagePrimaryKey): Promise<DeleteResult<PhpMessage>> {
throw Error("not implemented");
}
async list<ListOptions>(
// eslint-disable-next-line @typescript-eslint/no-unused-vars
pagination: Pagination,
// eslint-disable-next-line @typescript-eslint/no-unused-vars
options?: ListOptions,
// eslint-disable-next-line @typescript-eslint/no-unused-vars
context?: PhpMessageExecutionContext,
): Promise<ListResult<PhpMessage>> {
const findFilter: FindFilter = {
channel_id: context.channel_id,
parent_message_id: context.parent_message_id,
//id: context.id,
};
const list = await this.repository.find(findFilter, { pagination }, undefined);
return list;
}
}
@@ -0,0 +1,3 @@
import { Initializable, TwakeServiceProvider } from "../../../../core/platform/framework";
export interface PhpMessagesServiceAPI extends TwakeServiceProvider, Initializable {}
@@ -0,0 +1,11 @@
export const convertUuidV4ToV1 = (str: string): string => {
const uuid = [...str];
uuid[14] = "1";
return uuid.join("");
};
export const convertUuidV1ToV4 = (str: string): string => {
const uuid = [...str];
uuid[14] = "4";
return uuid.join("");
};
@@ -0,0 +1,45 @@
import { merge } from "lodash";
import {
Column,
Entity,
} from "../../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "workspace";
@Entity(TYPE, {
primaryKey: [["id"]],
type: TYPE,
})
export class PhpWorkspace {
@Column("id", "timeuuid")
id: string;
@Column("group_id", "timeuuid")
group_id: string;
@Column("name", "encoded_string")
name: string;
@Column("logo", "encoded_string")
logo: string;
@Column("stats", "encoded_string")
stats: string;
@Column("is_deleted", "boolean")
isDeleted: boolean;
@Column("is_archived", "boolean")
isArchived: boolean;
@Column("is_default", "boolean")
isDefault: boolean;
@Column("date_added", "number")
dateAdded: number;
}
export type PhpMessagePrimaryKey = Pick<PhpWorkspace, "id">;
export function getInstance(workspace: PhpWorkspace): PhpWorkspace {
return merge(new PhpWorkspace(), workspace);
}
@@ -0,0 +1,148 @@
import yargs from "yargs";
import twake from "../../../twake";
import ora from "ora";
import { TwakePlatform } from "../../../core/platform/platform";
import { DatabaseServiceAPI } from "../../../core/platform/services/database/api";
import { PhpWorkspace, TYPE as phpTYPE } from "./php-workspace/php-workspace-entity";
import { Pagination } from "../../../core/platform/framework/api/crud-service";
import Workspace, { TYPE, getInstance } from "../../../services/workspaces/entities/workspace";
import _ from "lodash";
import gr from "../../../services/global-resolver";
type Options = {
from?: string;
onlyCompany?: string;
onlyWorkspace?: string;
replaceExisting?: boolean;
};
class WorkspaceMigrator {
database: DatabaseServiceAPI;
constructor(readonly platform: TwakePlatform) {
this.database = this.platform.getProvider<DatabaseServiceAPI>("database");
}
public async run(options: Options = {}): Promise<void> {
const phpRepository = await this.database.getRepository(phpTYPE, PhpWorkspace);
const repository = await this.database.getRepository(TYPE, Workspace);
let waitForCompany = false;
if (options.from) {
waitForCompany = true;
}
// Get all companies
let page: Pagination = { limitStr: "100" };
// For each companies find workspaces
do {
const workspaceListResult = await phpRepository.find({}, { pagination: page }, undefined);
page = workspaceListResult.nextPage as Pagination;
for (const workspace of workspaceListResult.getEntities()) {
if (waitForCompany && options.from == `${workspace.group_id}`) {
waitForCompany = false;
}
if (!waitForCompany) {
if (
(!options.onlyCompany && !options.onlyWorkspace) ||
options.onlyCompany == `${workspace.group_id}`
) {
if (
!(await repository.findOne(
{ company_id: workspace.group_id, id: workspace.id },
{},
undefined,
)) ||
options.replaceExisting
) {
const newWorkspace = getInstance(
_.pick(
workspace,
"id",
"company_id",
"name",
"logo",
"stats",
"is_deleted",
"is_archived",
"is_default",
"date_added",
),
);
newWorkspace.company_id = workspace.group_id;
await repository.save(newWorkspace, undefined);
}
}
}
}
} while (page.page_token);
}
}
const services = [
"storage",
"counter",
"platform-services",
"applications",
"user",
"search",
"channels",
"database",
"webserver",
"message-queue",
"workspaces",
"console",
"auth",
];
const command: yargs.CommandModule<unknown, unknown> = {
command: "workspace",
describe: "command that allow you to migrate php workspaces to node",
builder: {
from: {
default: null,
type: "string",
description: "Start migration from this workspace ID",
},
onlyCompany: {
default: null,
type: "string",
description: "Migrate only this workspace ID",
},
onlyWorkspace: {
default: null,
type: "string",
description: "Migrate only this workspace ID",
},
replaceExisting: {
default: false,
type: "boolean",
description: "Replace already migrated workspaces",
},
},
// eslint-disable-next-line @typescript-eslint/no-unused-vars
handler: async argv => {
const spinner = ora({ text: "Migrating php worskpaces - " }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
const migrator = new WorkspaceMigrator(platform);
const from = argv.from as string | null;
const onlyCompany = argv.onlyCompany as string | null;
const onlyWorkspace = argv.onlyWorkspace as string | null;
const replaceExisting = (argv.replaceExisting || false) as boolean;
await migrator.run({
from,
onlyCompany,
onlyWorkspace,
replaceExisting,
});
return spinner.stop();
},
};
export default command;
+14
View File
@@ -0,0 +1,14 @@
import { CommandModule } from "yargs";
const command: CommandModule = {
describe: "Work with search middleware",
command: "search <command>",
builder: yargs =>
yargs.commandDir("search_cmds", {
visit: commandModule => commandModule.default,
}),
// eslint-disable-next-line @typescript-eslint/no-empty-function
handler: () => {},
};
export default command;
@@ -0,0 +1,160 @@
import yargs from "yargs";
import twake from "../../../twake";
import ora from "ora";
import { TwakePlatform } from "../../../core/platform/platform";
import { DatabaseServiceAPI } from "../../../core/platform/services/database/api";
import { Pagination } from "../../../core/platform/framework/api/crud-service";
import User, { TYPE as UserTYPE } from "../../../services/user/entities/user";
import { Channel } from "../../../services/channels/entities";
import Application, {
TYPE as ApplicationTYPE,
} from "../../../services/applications/entities/application";
import Repository from "../../../core/platform/services/database/services/orm/repository/repository";
import { SearchServiceAPI } from "../../../core/platform/services/search/api";
import CompanyUser, { TYPE as CompanyUserTYPE } from "../../../services/user/entities/company_user";
import { Message, TYPE as MessageTYPE } from "../../../services/messages/entities/messages";
import gr from "../../../services/global-resolver";
import {
MessageFile,
TYPE as MessageFileTYPE,
} from "../../../services/messages/entities/message-files";
type Options = {
repository?: string;
repairEntities?: boolean;
};
class SearchIndexAll {
database: DatabaseServiceAPI;
search: SearchServiceAPI;
constructor(readonly platform: TwakePlatform) {
this.database = this.platform.getProvider<DatabaseServiceAPI>("database");
this.search = this.platform.getProvider<SearchServiceAPI>("search");
}
public async run(options: Options = {}): Promise<void> {
const repositories: Map<string, Repository<any>> = new Map();
repositories.set("messages", await this.database.getRepository(MessageTYPE, Message));
repositories.set(
"message_files",
await this.database.getRepository(MessageFileTYPE, MessageFile),
);
repositories.set("users", await this.database.getRepository(UserTYPE, User));
repositories.set("channels", await this.database.getRepository("channels", Channel));
repositories.set(
"applications",
await this.database.getRepository(ApplicationTYPE, Application),
);
const repository = repositories.get(options.repository);
if (!repository) {
throw (
"No such repository ready for indexation, available are: " +
Array.from(repositories.keys()).join(", ")
);
}
// Complete user with companies in cache
if (options.repository === "users" && options.repairEntities) {
console.log("Complete user with companies in cache");
const companiesUsersRepository = await this.database.getRepository(
CompanyUserTYPE,
CompanyUser,
);
const userRepository = await this.database.getRepository(UserTYPE, User);
let page: Pagination = { limitStr: "100" };
// For each rows
do {
const list = await userRepository.find({}, { pagination: page }, undefined);
for (const user of list.getEntities()) {
const companies = await companiesUsersRepository.find(
{ user_id: user.id },
{},
undefined,
);
user.cache ||= { companies: [] };
user.cache.companies = companies.getEntities().map(company => company.group_id);
await repositories.get("users").save(user, undefined);
}
page = list.nextPage as Pagination;
await new Promise(r => setTimeout(r, 200));
} while (page.page_token);
}
console.log("Start indexing...");
let count = 0;
// Get all items
let page: Pagination = { limitStr: "100" };
do {
console.log("Indexed " + count + " items...");
const list = await repository.find({}, { pagination: page }, undefined);
page = list.nextPage as Pagination;
await this.search.upsert(list.getEntities());
count += list.getEntities().length;
await new Promise(r => setTimeout(r, 200));
} while (page.page_token);
console.log("Emptying flush (10s)...");
await new Promise(r => setTimeout(r, 10000));
console.log("Done!");
}
}
const services = [
"search",
"database",
"webserver",
"auth",
"counter",
"cron",
"message-queue",
"push",
"realtime",
"storage",
"tracker",
"websocket",
];
const command: yargs.CommandModule<unknown, unknown> = {
command: "index",
describe: "command to reindex search middleware from db entities",
builder: {
repository: {
default: "",
type: "string",
description: "Choose a repository to reindex",
},
repairEntities: {
default: false,
type: "boolean",
description: "Choose to repair entities too when possible",
},
},
// eslint-disable-next-line @typescript-eslint/no-unused-vars
handler: async argv => {
const spinner = ora({ text: "Reindex repository - " }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
const migrator = new SearchIndexAll(platform);
const repository = (argv.repository || "") as string;
if (!repository) {
throw "No repository was set.";
}
await migrator.run({
repository,
});
return spinner.stop();
},
};
export default command;
+14
View File
@@ -0,0 +1,14 @@
import { CommandModule } from "yargs";
const command: CommandModule = {
describe: "Manage Twake Users",
command: "users <command>",
builder: yargs =>
yargs.commandDir("users_cmds", {
visit: commandModule => commandModule.default,
}),
// eslint-disable-next-line @typescript-eslint/no-empty-function
handler: () => {},
};
export default command;
@@ -0,0 +1,85 @@
import yargs from "yargs";
import ora from "ora";
import twake from "../../../twake";
import Table from "cli-table";
import { exit } from "process";
import gr from "../../../services/global-resolver";
/**
* Merge command parameters. Check the builder definition below for more details.
*/
type CLIArgs = {
id: string;
};
const services = [
"storage",
"counter",
"message-queue",
"platform-services",
"user",
"search",
"database",
"webserver",
"statistics",
"applications",
"auth",
"realtime",
"websocket",
];
const command: yargs.CommandModule<unknown, CLIArgs> = {
command: "remove",
describe: "command that allow you to remove one user",
builder: {
id: {
default: "",
type: "string",
description: "User ID",
},
},
handler: async argv => {
const tableBefore = new Table({
head: ["User ID", "Username", "Deleted"],
colWidths: [40, 40, 10],
});
const tableAfter = new Table({
head: ["User ID", "Username", "Deleted"],
colWidths: [40, 40, 10],
});
const spinner = ora({ text: "Retrieving user" }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
const user = await gr.services.users.get({ id: argv.id });
if (!user) {
console.error("Error: You need to provide User ID");
return spinner.stop();
}
if (user) {
// Table before
tableBefore.push([user.id, user.username_canonical, user.deleted]);
await gr.services.users.anonymizeAndDelete(
{ id: user.id },
{
user: { id: user.id, server_request: true },
},
);
const finalUser = await gr.services.users.get({ id: argv.id });
// Table after
tableAfter.push([finalUser.id, finalUser.username_canonical, finalUser.deleted]);
spinner.stop();
}
exit();
},
};
export default command;
@@ -0,0 +1,14 @@
import { CommandModule } from "yargs";
const command: CommandModule = {
describe: "Manage Twake Workspaces",
command: "workspace <command>",
builder: yargs =>
yargs.commandDir("workspace_cmds", {
visit: commandModule => commandModule.default,
}),
// eslint-disable-next-line @typescript-eslint/no-empty-function
handler: () => {},
};
export default command;
@@ -0,0 +1,47 @@
import yargs from "yargs";
import ora from "ora";
import Table from "cli-table";
import twake from "../../../twake";
import gr from "../../../services/global-resolver";
/**
* Merge command parameters. Check the builder definition below for more details.
*/
type ListParams = {
size: string;
};
const services = [
"platform-services",
"user",
"search",
"channels",
"notifications",
"database",
"webserver",
"message-queue",
];
const command: yargs.CommandModule<ListParams, ListParams> = {
command: "list",
describe: "List Twake workspaces",
builder: {
size: {
default: "50",
type: "string",
description: "Number of workspaces to fetch",
},
},
handler: async argv => {
const table = new Table({ head: ["ID", "Name"], colWidths: [40, 50] });
const spinner = ora({ text: "List Twake workspaces" }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
const workspaces = await gr.services.workspaces.list({ limitStr: argv.size });
spinner.stop();
workspaces.getEntities().forEach(ws => table.push([ws.id, ws.name]));
console.log(table.toString());
},
};
export default command;
@@ -0,0 +1,49 @@
import yargs from "yargs";
import ora from "ora";
import Table from "cli-table";
import twake from "../../../twake";
import gr from "../../../services/global-resolver";
/**
* Merge command parameters. Check the builder definition below for more details.
*/
type ListParams = {
id: string;
};
const services = [
"platform-services",
"user",
"search",
"channels",
"notifications",
"database",
"webserver",
"message-queue",
];
const command: yargs.CommandModule<ListParams, ListParams> = {
command: "user",
describe: "List workspace ers",
builder: {
id: {
default: "f339d54a-e833-11ea-92c3-0242ac120004",
type: "string",
description: "Workspace ID",
},
},
handler: async argv => {
const table = new Table({ head: ["user ID", "Date Added"], colWidths: [40, 40] });
const spinner = ora({ text: "Retrieving workspace users" }).start();
const platform = await twake.run(services);
await gr.doInit(platform);
const users = await gr.services.workspaces.getUsers({ workspaceId: argv.id });
spinner.stop();
users
.getEntities()
.forEach(u => table.push([u.id, new Date(u.dateAdded).toLocaleDateString()]));
console.log(table.toString());
},
};
export default command;
+28
View File
@@ -0,0 +1,28 @@
import yargs from "yargs";
import { logger } from "../core/platform/framework/logger";
process.env.NODE_ENV = "cli";
yargs
.strict()
.usage("Usage: $0 <command> [options]")
.middleware([
argv => {
logger.level = argv.verbose ? "debug" : "fatal";
},
])
.commandDir("cmds", {
visit: commandModule => commandModule.default,
})
.option("verbose", {
alias: "v",
default: false,
type: "boolean",
description: "Run with verbose logging",
})
.demandCommand(1, "Please supply a valid command")
.alias("help", "h")
.help("help")
.version()
.epilogue("for more information, go to https://twake.app")
.example("$0 <command> --help", "show help of the issue command").argv;
@@ -0,0 +1,2 @@
node_modules
yarn.lock
@@ -0,0 +1,4 @@
Edit the file to your needs
yarn install
node index.js
@@ -0,0 +1,214 @@
/* eslint-disable */
var cassandra = require("cassandra-driver");
var fromAuthProvider = new cassandra.auth.PlainTextAuthProvider("", "");
var fromContactPoints = [""];
var fromKeyspace = "twake";
var toAuthProvider = new cassandra.auth.PlainTextAuthProvider("", "");
var toContactPoints = [""];
var toKeyspace = "twake";
var forceUpdateAll = false;
var ignoreTables = ["notification"];
var countersTables = ["statistics", "stats_counter", "channel_counters", "scheduled_queue_counter"];
var specialConversions = {
"twake.group_user.date_added": value => {
return new Date(value).getTime();
},
};
// -- start process
var fromClient = new cassandra.Client({
localDataCenter: "datacenter1",
contactPoints: fromContactPoints,
authProvider: fromAuthProvider,
keyspace: fromKeyspace,
});
var toClient = new cassandra.Client({
localDataCenter: "datacenter1",
contactPoints: toContactPoints,
authProvider: toAuthProvider,
keyspace: toKeyspace,
queryOptions: {
consistency: cassandra.types.consistencies.quorum,
},
});
// -- Get all tables and copy schema
async function client(origin, query, parameters, options) {
return await new Promise((resolve, reject) => {
origin.execute(query, [], {}, function (err, result) {
if (err) {
reject({ err, result });
} else {
resolve(result);
}
});
});
}
(async () => {
const result = await client(
fromClient,
"SELECT table_name from system_schema.tables WHERE keyspace_name = '" + fromKeyspace + "'",
[],
{},
);
for (row of result.rows) {
try {
const fromTable = fromKeyspace + "." + row.table_name;
const toTable = toKeyspace + "." + row.table_name;
if (ignoreTables.includes(row.table_name)) {
console.log(fromTable.padEnd(50) + " | " + "ignored".padEnd(20) + " | ⏺");
continue;
}
let fromCount = 0;
const destColumns = (
await client(
toClient,
"SELECT column_name from system_schema.columns where keyspace_name = '" +
toKeyspace +
"' and table_name = '" +
row.table_name +
"'",
[],
{},
)
).rows.map(r => r.column_name);
try {
const fromResult = await client(
fromClient,
"SELECT count(*) from " + fromTable + "",
[],
{},
);
fromCount = parseInt(fromResult.rows[0].count);
} catch (err) {
fromCount = NaN;
}
if (destColumns.length === 0) {
if (fromCount > 0)
console.log(
fromTable.padEnd(50) + " | " + ("not_in_dest" + "/" + fromCount).padEnd(20) + " | ⏺",
);
continue;
}
try {
const toResult = await client(toClient, "SELECT count(*) from " + toTable + "", [], {});
const toCount = parseInt(toResult.rows[0].count);
console.log(
fromTable.padEnd(50) +
" | " +
(toCount + "/" + fromCount).padEnd(20) +
" | " +
(toCount >= fromCount ? "✅" : "❌"),
);
if (row.table_name.indexOf("counter") >= 0 || countersTables.includes(row.table_name)) {
console.log(
fromTable.padEnd(50) + " | " + ("counter_table" + "/" + fromCount).padEnd(20) + " | 🧮",
);
if (fromCount > toCount || !fromCount || forceUpdateAll) {
await new Promise(r => {
fromClient.eachRow(
"SELECT JSON * from " + fromTable,
[],
{ prepare: true, fetchSize: 1000 },
async function (n, row) {
//
//TODO handle counters (it is special !)
},
async function (err, result) {
if (result && result.nextPage) {
await new Promise(r => setTimeout(r, 1000));
result.nextPage();
} else {
r();
}
},
);
});
}
continue;
}
if (fromCount > toCount || !fromCount || forceUpdateAll) {
await new Promise(r => {
fromClient.eachRow(
"SELECT JSON * from " + fromTable,
[],
{ prepare: true, fetchSize: 1000 },
async function (n, row) {
try {
const json = JSON.parse(row["[json]"]);
//The from table can have additional depreciated fields, we need to remove them
const filteredJson = {};
for (const col of destColumns) {
if (
typeof json[col] == "string" &&
(json[col] || "").match(
/[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}.[0-9]+/,
)
) {
json[col] = json[col].split(".")[0];
}
if (specialConversions[fromTable + "." + col]) {
json[col] = specialConversions[fromTable + "." + col](json[col]);
}
if (json[col] !== undefined) filteredJson[col] = json[col];
}
await client(
toClient,
"INSERT INTO " +
toTable +
" JSON '" +
JSON.stringify(filteredJson).replace(/'/g, "'$&") +
"'",
[],
{},
);
} catch (err) {
console.log(err);
}
},
async function (err, result) {
if (result && result.nextPage) {
await new Promise(r => setTimeout(r, 1000));
result.nextPage();
} else {
r();
}
},
);
});
}
} catch (err) {
console.log(
fromTable.padEnd(50) + " | " + ("error" + "/" + fromCount).padEnd(20) + " | ❌",
);
}
} catch (err) {
console.log(err);
continue;
}
//TODO copy content
}
})();
@@ -0,0 +1,9 @@
{
"name": "copy-cluster",
"version": "1.0.0",
"main": "index.js",
"license": "MIT",
"dependencies": {
"cassandra-driver": "^4.6.3"
}
}
@@ -0,0 +1,2 @@
node_modules
yarn.lock
@@ -0,0 +1,4 @@
Edit the file to your needs
yarn install
node index.js
@@ -0,0 +1,98 @@
/* eslint-disable */
var cassandra = require("cassandra-driver");
var fromAuthProvider = new cassandra.auth.PlainTextAuthProvider("", "");
var fromContactPoints = [""];
var toAuthProvider = new cassandra.auth.PlainTextAuthProvider("", "");
var toContactPoints = [""];
// -- start process
var fromClient = new cassandra.Client({
localDataCenter: "datacenter1",
contactPoints: fromContactPoints,
authProvider: fromAuthProvider,
keyspace: "twake",
});
var toClient = new cassandra.Client({
localDataCenter: "datacenter1",
contactPoints: toContactPoints,
authProvider: toAuthProvider,
keyspace: "twake",
queryOptions: {
consistency: cassandra.types.consistencies.quorum,
},
});
// -- Get all tables and copy schema
async function client(origin, query, parameters, options) {
return await new Promise((resolve, reject) => {
origin.execute(query, [], {}, function (err, result) {
if (err) {
reject({ err, result });
} else {
resolve(result);
}
});
});
}
// Get all threads and then copy all messages in each threads
let copiedMessages = 0;
let copiedThreads = 0;
(async () => {
await new Promise(r => {
fromClient.eachRow(
"SELECT id from twake.threads",
[],
{ prepare: true, fetchSize: 50 },
async function (n, row) {
const threadId = row["id"];
console.log("Threads / Messages :", copiedThreads, "/", copiedMessages);
copiedThreads++;
await new Promise(r2 => {
fromClient.eachRow(
"SELECT JSON * from twake.messages where thread_id = ? order by id desc",
[threadId],
{ prepare: true, fetchSize: 1000 },
async function (n, row) {
const message = row["[json]"];
copiedMessages++;
await toClient.execute(
"INSERT INTO twake.messages JSON '" + message.replace(/'/g, "'$&") + "'",
[],
{ prepare: true },
);
},
async function (err, result) {
if (result && result.nextPage) {
await new Promise(r => setTimeout(r, 100));
result.nextPage();
} else {
r2();
}
},
);
});
},
async function (err, result) {
if (result && result.nextPage) {
await new Promise(r => setTimeout(r, 2000));
result.nextPage();
} else {
r();
}
},
);
});
})();
@@ -0,0 +1,9 @@
{
"name": "copy-messages",
"version": "1.0.0",
"main": "index.js",
"license": "MIT",
"dependencies": {
"cassandra-driver": "^4.6.3"
}
}
@@ -0,0 +1,4 @@
Edit the file to your needs
yarn install
node index.js
@@ -0,0 +1,88 @@
/* eslint-disable */
var cassandra = require("cassandra-driver");
var authProvider = new cassandra.auth.PlainTextAuthProvider("", "");
var contactPoints = [""];
var from = "table_a";
var to = "table_b";
var client = new cassandra.Client({
localDataCenter: "datacenter1",
contactPoints: contactPoints,
authProvider: authProvider,
keyspace: "twake",
});
//Copy object to other table
(async () => {
const query = "SELECT * FROM " + from;
const parameters = [];
let count = 0;
let max = 0;
let min = new Date().getTime();
let rows = [];
const options = { prepare: true, fetchSize: 1000 };
await new Promise(r => {
client.eachRow(
query,
parameters,
options,
function (n, row) {
min = Math.min(parseInt(row.created_at.toString()), min);
max = Math.max(parseInt(row.created_at.toString()), max);
count++;
if (count % 100 == 0) {
console.log(count);
}
rows.push(row);
},
function (err, result) {
if (result && result.nextPage) {
result.nextPage();
} else {
r();
}
},
);
});
console.log("Downloaded ", count, " rows of ", from, ". Starting copy...");
let copyCount = 0;
for (let i = 0; i < count; i++) {
copyCount++;
if (copyCount % 100 == 0) {
console.log(copyCount, "of", count);
}
const row = rows[i];
await new Promise(r => {
let query =
"UPDATE " +
to +
" SET answers=?, created_at=?, created_by=?, last_activity=?, participants=?, updated_at=? WHERE id=?";
client.execute(
query,
[
row.answers,
row.created_at,
row.created_by,
row.last_activity,
row.participants,
row.updated_at,
row.id,
],
() => {
r();
},
);
});
}
console.log("Ended with ", count, "threads");
})();
@@ -0,0 +1,6 @@
{
"name": "copy-table",
"version": "1.0.0",
"main": "index.js",
"license": "MIT"
}
@@ -0,0 +1,4 @@
Edit the file to your needs
yarn install
node index.js
@@ -0,0 +1,96 @@
/* eslint-disable */
var cassandra = require("cassandra-driver");
var authProvider = new cassandra.auth.PlainTextAuthProvider("", "");
var contactPoints = [""];
var client = new cassandra.Client({
localDataCenter: "datacenter1",
contactPoints: contactPoints,
authProvider: authProvider,
keyspace: "twake",
});
var threadsTable = "threads";
var messagesTable = "messages";
//Copy object to other table
(async () => {
const query = "SELECT * FROM " + threadsTable + " where answers>1000 allow filtering";
const parameters = [];
const options = { prepare: true, fetchSize: 1000 };
let count = 0;
let threads = {};
await new Promise(r => {
client.eachRow(
query,
parameters,
options,
function (n, row) {
threads[row.id] ||= { answers: 0, id: row.id };
count++;
},
function (err, result) {
if (result && result.nextPage) {
result.nextPage();
} else {
r();
}
},
);
});
console.log(
"Downloaded ",
Object.keys(threads).length,
" threads of ",
messagesTable,
". Starting copy...",
);
for (var key of Object.keys(threads)) {
await new Promise(r => {
const query = "SELECT id FROM " + messagesTable + " where thread_id=" + key;
const parameters = [];
const options = { prepare: true, fetchSize: 1000 };
client.eachRow(
query,
parameters,
options,
function (n, row) {
threads[key].answers++;
},
function (err, result) {
if (result && result.nextPage) {
result.nextPage();
} else {
r();
}
},
);
r();
});
}
var copyCount = 0;
for (var key of Object.keys(threads)) {
copyCount++;
if (copyCount % 100 == 0) {
console.log(copyCount, "of", Object.keys(threads));
}
const row = threads[key];
await new Promise(r => {
let query = "UPDATE " + threadsTable + " SET answers=? WHERE id=?";
console.log(query, [row.answers, row.id + ""]);
//client.execute(query, [row.answers, row.id], () => {
// r();
//});
});
}
console.log("Ended with ", Object.keys(threads), "threads");
})();
@@ -0,0 +1,9 @@
{
"name": "repair-threads",
"version": "1.0.0",
"main": "index.js",
"license": "MIT",
"dependencies": {
"cassandra-driver": "^4.6.3"
}
}
@@ -0,0 +1,3 @@
import config from "config";
export default config;
@@ -0,0 +1,50 @@
import legacy from "./legacy";
import v1 from "./v1";
import v2 from "./v2";
import { createHash } from "crypto";
export type CryptoResult = {
/**
* Encrypted | Decrypted data if done, original data if not done due to error or not encrypted data to decrypt
*/
data: any;
/**
* Is there an error while processing the input data?
*/
err?: Error;
/**
* Did we encrypted | decrypted the data? true if yes.
*/
done?: boolean;
};
export function decrypt(data: string, encryptionKey: string): CryptoResult {
let result = v2.decrypt(data, encryptionKey);
if (result.done) {
return result;
}
result = v1.decrypt(data, encryptionKey);
if (result.done) {
return result;
}
result = legacy.decrypt(data, encryptionKey);
if (result.done) {
return result;
}
return result;
}
export function md5(value: string): string {
return createHash("md5").update(value).digest("hex");
}
export function encrypt(
value: any,
encryptionKey: any,
options: { disableSalts?: boolean } = {},
): CryptoResult {
return v2.encrypt(value, encryptionKey, options);
}
@@ -0,0 +1,45 @@
import { createDecipheriv } from "crypto";
import { CryptoResult } from ".";
const PREFIX = "encrypted_";
const DEFAULT_IV = "twake_constantiv";
export default {
decrypt,
encrypt,
};
function decrypt(data: string, encryptionKey: string): CryptoResult {
if (!data || !data.startsWith(PREFIX)) {
return {
data,
done: false,
};
}
const toDecode = data.substr(PREFIX.length);
const encryptedArray = toDecode.split("_");
// If array length is 1, the value has not been salted and iv is the default one
const encrypted = Buffer.from(encryptedArray[0], "base64");
const salt = encryptedArray[1]
? Buffer.from(encryptedArray[1], "utf-8")
: Buffer.from("", "utf-8");
const iv = encryptedArray[2]
? Buffer.from(encryptedArray[2], "base64")
: Buffer.from(DEFAULT_IV, "utf-8");
const password = Buffer.concat([Buffer.from(encryptionKey, "hex"), salt], 32);
const decipher = createDecipheriv("AES-256-CBC", password, iv);
return {
data: Buffer.concat([decipher.update(encrypted), decipher.final()]).toString("utf-8"),
done: true,
};
}
function encrypt(data: any): CryptoResult {
return {
data,
done: false,
};
}
+67
View File
@@ -0,0 +1,67 @@
import { createCipheriv, createDecipheriv, createHash, randomBytes } from "crypto";
import { CryptoResult } from ".";
export default {
encrypt,
decrypt,
};
function encrypt(
data: any,
encryptionKey: any,
options: { disableSalts?: boolean } = {},
): CryptoResult {
const key = createHash("sha256").update(String(encryptionKey)).digest("base64").substr(0, 32);
try {
const iv = options.disableSalts ? "0000000000000000" : randomBytes(16);
const cipher = createCipheriv("aes-256-cbc", key, iv);
const encrypted = Buffer.concat([cipher.update(JSON.stringify(data)), cipher.final()]).toString(
"hex",
);
return {
data: `${iv.toString("hex")}:${encrypted}`,
done: true,
};
} catch (err) {
return {
err,
data,
done: false,
};
}
}
function decrypt(data: string, encryptionKey: any): CryptoResult {
const key = createHash("sha256").update(String(encryptionKey)).digest("base64").substr(0, 32);
const encryptedArray = data.split(":");
if (!encryptedArray.length || encryptedArray.length !== 2) {
return {
data,
done: false,
};
}
let iv: Buffer | string = Buffer.from(encryptedArray[0], "hex");
if (encryptedArray[0] === "0000000000000000") {
iv = "0000000000000000";
}
try {
const encrypted = Buffer.from(encryptedArray[1], "hex");
const decipher = createDecipheriv("aes-256-cbc", key, iv);
const decrypt = JSON.parse(
Buffer.concat([decipher.update(encrypted), decipher.final()]).toString(),
);
return {
data: decrypt,
done: true,
};
} catch (err) {
return {
data,
done: false,
};
}
}
+70
View File
@@ -0,0 +1,70 @@
import { createCipheriv, createDecipheriv, createHash, randomBytes } from "crypto";
import { CryptoResult } from ".";
export default {
encrypt,
decrypt,
};
function encrypt(
data: any,
encryptionKey: any,
options: { disableSalts?: boolean } = {},
): CryptoResult {
const key = createHash("sha256").update(String(encryptionKey)).digest("hex").substr(0, 32);
try {
const iv = options.disableSalts ? "0000000000000000" : randomBytes(16);
const cipher = createCipheriv("aes-256-gcm", key, iv);
const encrypted =
cipher.update(JSON.stringify(data), "utf8", "base64") + cipher.final("base64");
return {
data: `${iv.toString("base64")}:${cipher.getAuthTag().toString("base64")}:${encrypted}`,
done: true,
};
} catch (err) {
return {
err,
data,
done: false,
};
}
}
function decrypt(data: string, encryptionKey: any): CryptoResult {
const key = createHash("sha256").update(String(encryptionKey)).digest("hex").substr(0, 32);
const encryptedArray = data.split(":");
if (!encryptedArray.length || encryptedArray.length !== 3) {
return {
data,
done: false,
};
}
let iv: Buffer | string = Buffer.from(encryptedArray[0], "base64");
if (encryptedArray[0] === "0000000000000000") {
iv = "0000000000000000";
}
try {
const decipher = createDecipheriv("aes-256-gcm", key, iv);
decipher.setAuthTag(Buffer.from(encryptedArray[1], "base64"));
const encrypted = encryptedArray[2];
let str = decipher.update(encrypted, "base64", "utf8");
str += decipher.final("utf8");
const decrypt = JSON.parse(str);
return {
data: decrypt,
done: true,
};
} catch (err) {
return {
data,
done: false,
};
}
}
@@ -0,0 +1,7 @@
import { TwakeServiceOptions } from "./service-options";
import { TwakeServiceConfiguration } from "./service-configuration";
export class TwakeAppConfiguration extends TwakeServiceOptions<TwakeServiceConfiguration> {
services: Array<string>;
servicesPath: string;
}
@@ -0,0 +1,2 @@
// eslint-disable-next-line @typescript-eslint/no-explicit-any
export type Class = { new (...args: any[]): any };
@@ -0,0 +1,67 @@
import { BehaviorSubject } from "rxjs";
import { TwakeService } from "./service";
import { TwakeServiceProvider } from "./service-provider";
import { ServiceDefinition } from "./service-definition";
import { TwakeServiceState } from "./service-state";
import { logger } from "../logger";
export class TwakeComponent {
instance: TwakeService<TwakeServiceProvider>;
components: Array<TwakeComponent> = new Array<TwakeComponent>();
constructor(public name: string, private serviceDefinition: ServiceDefinition) {}
getServiceDefinition(): ServiceDefinition {
return this.serviceDefinition;
}
setServiceInstance(instance: TwakeService<TwakeServiceProvider>): void {
this.instance = instance;
}
getServiceInstance(): TwakeService<TwakeServiceProvider> {
return this.instance;
}
addDependency(component: TwakeComponent): void {
this.components.push(component);
}
getStateTree(): string {
return `${this.name}(${this.instance.state.value}) => {${this.components
.map(component => component.getStateTree())
.join(",")}}`;
}
async switchToState(
state: TwakeServiceState.Initialized | TwakeServiceState.Started | TwakeServiceState.Stopped,
recursionDepth?: number,
): Promise<void> {
if (recursionDepth > 10) {
logger.error("Maximum recursion depth exceeded (will exit process)");
process.exit(1);
}
const states: BehaviorSubject<TwakeServiceState>[] = this.components.map(
component => component.instance.state,
);
if (states.length) {
for (const component of this.components) {
await component.switchToState(state, (recursionDepth || 0) + 1);
}
logger.info(`Children of ${this.name} are all in ${state} state`);
logger.info(this.getStateTree());
} else {
logger.info(`${this.name} does not have children`);
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
async function _switchServiceToState(service: TwakeService<any>) {
state === TwakeServiceState.Initialized && (await service.init());
state === TwakeServiceState.Started && (await service.start());
state === TwakeServiceState.Stopped && (await service.stop());
}
await _switchServiceToState(this.instance);
}
}
@@ -0,0 +1,3 @@
export const PREFIX_METADATA = "service:prefix";
export const CONSUMES_METADATA = "service:consumes";
export const SERVICENAME_METADATA = "service:name";
@@ -0,0 +1,73 @@
import { TwakeAppConfiguration } from "./application-configuration";
import { TwakeComponent } from "./component";
import { TwakeContext } from "./context";
import { TwakeServiceProvider } from "./service-provider";
import { TwakeService } from "./service";
import { TwakeServiceState } from "./service-state";
import * as ComponentUtils from "../utils/component-utils";
/**
* A container contains components. It provides methods to manage and to retrieve them.
*/
export abstract class TwakeContainer
extends TwakeService<TwakeServiceProvider>
implements TwakeContext
{
private components: Map<string, TwakeComponent>;
name = "TwakeContainer";
constructor(protected options?: TwakeAppConfiguration) {
super(options);
}
abstract loadComponents(): Promise<Map<string, TwakeComponent>>;
abstract loadComponent(name: string): Promise<TwakeComponent>;
getProvider<T extends TwakeServiceProvider>(name: string): T {
const service = this.components.get(name)?.getServiceInstance();
if (!service) {
throw new Error(`Service "${name}" not found`);
}
return service.api() as T;
}
async doInit(): Promise<this> {
this.components = await this.loadComponents();
await ComponentUtils.buildDependenciesTree(this.components, async (name: string) => {
const component = await this.loadComponent(name);
if (component) this.components.set(name, component);
return component;
});
await this.launchInit();
return this;
}
protected async launchInit(): Promise<this> {
await this.switchToState(TwakeServiceState.Initialized);
return this;
}
async doStart(): Promise<this> {
await this.switchToState(TwakeServiceState.Started);
return this;
}
async doStop(): Promise<this> {
await this.switchToState(TwakeServiceState.Stopped);
return this;
}
protected async switchToState(
state: TwakeServiceState.Started | TwakeServiceState.Initialized | TwakeServiceState.Stopped,
): Promise<void> {
await ComponentUtils.switchComponentsToState(this.components, state);
}
}
@@ -0,0 +1,5 @@
import { TwakeServiceProvider } from "./service-provider";
export interface TwakeContext {
getProvider<T extends TwakeServiceProvider>(name: string): T;
}
@@ -0,0 +1,233 @@
import { User } from "../../../../utils/types";
export class ContextualizedTarget {
context?: ExecutionContext;
readonly operation: OperationType;
}
export class EntityTarget<Entity> extends ContextualizedTarget {
/**
*
* @param type type of entity
* @param entity the entity itself
*/
constructor(readonly type: string, readonly entity: Entity) {
super();
}
}
export class UpdateResult<Entity> extends EntityTarget<Entity> {
readonly operation = OperationType.UPDATE;
/**
* Result sent back by the underlying database
*/
// eslint-disable-next-line @typescript-eslint/no-explicit-any
raw?: any;
/**
* Number of rows affected by the update.
*/
affected?: number;
}
export class CreateResult<Entity> extends EntityTarget<Entity> {
readonly operation = OperationType.CREATE;
/**
* Result sent back by the underlying database
*/
// eslint-disable-next-line @typescript-eslint/no-explicit-any
raw?: any;
}
export class SaveResult<Entity> extends EntityTarget<Entity> {
/**
* Result sent back by the underlying database
*/
// eslint-disable-next-line @typescript-eslint/no-explicit-any
raw?: any;
/**
*
* @param type Type of entity
* @param entity The entity itself
* @param operation Save can be for a "create", an "update" or "exists" when resource already exists
*/
constructor(
readonly type: string,
readonly entity: Entity,
readonly operation: OperationType.UPDATE | OperationType.CREATE | OperationType.EXISTS,
) {
super(type, entity);
}
}
export class DeleteResult<Entity> extends EntityTarget<Entity> {
readonly operation = OperationType.DELETE;
/**
*
* @param type type of entity
* @param entity the entity itself
* @param deleted the entity has been deleted or not
*/
constructor(readonly type: string, readonly entity: Entity, readonly deleted: boolean) {
super(type, entity);
}
}
export class ListResult<Entity> extends ContextualizedTarget implements Paginable {
// next page token
page_token: string;
constructor(readonly type: string, protected entities: Entity[], readonly nextPage?: Paginable) {
super();
this.page_token = nextPage?.page_token;
}
mapEntities(mapper: <T extends Entity>(entity: Entity) => T): void {
this.entities = this.entities.map(entity => mapper(entity));
}
filterEntities(filter: (entity: Entity) => boolean): void {
this.entities = this.entities.filter(entity => filter(entity));
}
getEntities(): Entity[] {
return this.entities || [];
}
isEmpty(): boolean {
return this.getEntities().length === 0;
}
}
export enum OperationType {
CREATE = "create",
UPDATE = "update",
SAVE = "save",
DELETE = "delete",
EXISTS = "exists",
}
export declare type EntityOperationResult<Entity> =
| CreateResult<Entity>
| UpdateResult<Entity>
| SaveResult<Entity>
| DeleteResult<Entity>;
export interface ExecutionContext {
user: User;
reqId?: string;
url?: string;
method?: string;
transport?: "http" | "ws";
}
export class CrudException extends Error {
constructor(readonly details: string, readonly status: number) {
super();
this.message = details;
}
static badRequest(details: string): CrudException {
return new CrudException(details, 400);
}
static notFound(details: string): CrudException {
return new CrudException(details, 404);
}
static forbidden(details: string): CrudException {
return new CrudException(details, 403);
}
static notImplemented(details: string): CrudException {
return new CrudException(details, 501);
}
static badGateway(details: string): CrudException {
return new CrudException(details, 502);
}
}
export interface Paginable {
page_token?: string;
limitStr?: string;
reversed?: boolean;
}
export class Pagination implements Paginable {
reversed?: boolean;
constructor(readonly page_token?: string, readonly limitStr = "100", reversed = false) {
this.reversed = reversed;
}
public static fromPaginable(p: Paginable) {
return new Pagination(p.page_token, p.limitStr, p.reversed);
}
}
export interface CRUDService<Entity, PrimaryKey, Context extends ExecutionContext> {
/**
* Creates a resource
*
* @param item
* @param context
*/
create?(item: Entity, context?: Context): Promise<CreateResult<Entity>>;
/**
* Get a resource
*
* @param pk
* @param context
*/
get(pk: PrimaryKey, context?: Context): Promise<Entity>;
/**
* Update a resource
*
* @param pk
* @param item
* @param context
*/
update?(
pk: PrimaryKey,
item: Entity,
context?: Context /* TODO: Options */,
): Promise<UpdateResult<Entity>>;
/**
* Save a resource.
* If the resource exists, it is updated, if it does not exists, it is created.
*
* @param itemOrItems
* @param options
* @param context
*/
save?<SaveOptions>(
item: Entity,
options?: SaveOptions,
context?: Context,
): Promise<SaveResult<Entity>>;
/**
* Delete a resource
*
* @param pk
* @param context
*/
delete(pk: PrimaryKey, context?: Context): Promise<DeleteResult<Entity>>;
/**
* List a resource
*
* @param context
*/
list<ListOptions>(
pagination: Paginable,
options?: ListOptions,
context?: Context /* TODO: Options */,
): Promise<ListResult<Entity>>;
}
@@ -0,0 +1,14 @@
export * from "./application-configuration";
export * from "./class";
export * from "./component";
export * from "./constants";
export * from "./context";
export * from "./container";
export * from "./lifecycle";
export * from "./service-configuration";
export * from "./service-definition";
export * from "./service-interface";
export * from "./service-options";
export * from "./service-provider";
export * from "./service-state";
export * from "./service";
@@ -0,0 +1,6 @@
import { TwakeContext } from "./context";
export interface Initializable {
// TODO: Flag to tell if a service which fails to initialize must break all
init?(context?: TwakeContext): Promise<this>;
}
@@ -0,0 +1,3 @@
export interface TwakeServiceConfiguration {
get<T>(name?: string, defaultValue?: T): T;
}
@@ -0,0 +1,6 @@
import { Class } from "./class";
export interface ServiceDefinition {
name: string;
clazz: Class;
}
@@ -0,0 +1,8 @@
import { TwakeServiceProvider } from "./service-provider";
export interface TwakeServiceInterface<T extends TwakeServiceProvider> {
doInit(): Promise<this>;
doStart(): Promise<this>;
doStop(): Promise<this>;
api(): T;
}
@@ -0,0 +1,5 @@
export class TwakeServiceOptions<TwakeServiceConfiguration> {
name?: string;
// TODO: configuration is abstract and comes from all others
configuration?: TwakeServiceConfiguration;
}
@@ -0,0 +1,3 @@
export interface TwakeServiceProvider {
version: string;
}
@@ -0,0 +1,10 @@
export enum TwakeServiceState {
Ready = "READY",
Initializing = "INITIALIZING",
Initialized = "INITIALIZED",
Starting = "STARTING",
Started = "STARTED",
Stopping = "STOPPING",
Stopped = "STOPPED",
Errored = "ERRORED",
}
@@ -0,0 +1,133 @@
import { BehaviorSubject } from "rxjs";
import { TwakeServiceInterface } from "./service-interface";
import { TwakeServiceProvider } from "./service-provider";
import { TwakeServiceState } from "./service-state";
import { TwakeServiceConfiguration } from "./service-configuration";
import { TwakeContext } from "./context";
import { TwakeServiceOptions } from "./service-options";
import { CONSUMES_METADATA, PREFIX_METADATA } from "./constants";
import { getLogger, logger } from "../logger";
import { TwakeLogger } from "..";
const pendingServices: any = {};
export abstract class TwakeService<T extends TwakeServiceProvider>
implements TwakeServiceInterface<TwakeServiceProvider>
{
state: BehaviorSubject<TwakeServiceState>;
readonly name: string;
protected readonly configuration: TwakeServiceConfiguration;
context: TwakeContext;
logger: TwakeLogger;
constructor(protected options?: TwakeServiceOptions<TwakeServiceConfiguration>) {
this.state = new BehaviorSubject<TwakeServiceState>(TwakeServiceState.Ready);
// REMOVE ME, we should import config from framework folder instead
this.configuration = options?.configuration;
this.logger = getLogger(`core.platform.services.${this.name}Service`);
}
abstract api(): T;
public get prefix(): string {
return Reflect.getMetadata(PREFIX_METADATA, this) || "/";
}
getConsumes(): Array<string> {
return Reflect.getMetadata(CONSUMES_METADATA, this) || [];
}
async init(): Promise<this> {
if (this.state.value !== TwakeServiceState.Ready) {
logger.info("Service %s is already initialized", this.name);
return this;
}
try {
logger.info("Initializing service %s", this.name);
pendingServices[this.name] = true;
this.state.next(TwakeServiceState.Initializing);
await this.doInit();
this.state.next(TwakeServiceState.Initialized);
logger.info("Service %s is initialized", this.name);
delete pendingServices[this.name];
logger.info("Pending services: %s", JSON.stringify(Object.keys(pendingServices)));
return this;
} catch (err) {
logger.error("Error while initializing service %s", this.name);
logger.error(err);
this.state.error(new Error(`Error while initializing service ${this.name}`));
throw err;
}
}
async doInit(): Promise<this> {
return this;
}
async doStart(): Promise<this> {
return this;
}
async start(): Promise<this> {
if (
this.state.value === TwakeServiceState.Starting ||
this.state.value === TwakeServiceState.Started
) {
logger.info("Service %s is already started", this.name);
return this;
}
try {
logger.info("Starting service %s", this.name);
this.state.next(TwakeServiceState.Starting);
await this.doStart();
this.state.next(TwakeServiceState.Started);
logger.info("Service %s is started", this.name);
return this;
} catch (err) {
logger.error("Error while starting service %s", this.name, err);
logger.error(err);
this.state.error(new Error(`Error while starting service ${this.name}`));
throw err;
}
}
async stop(): Promise<this> {
if (
this.state.value === TwakeServiceState.Stopping ||
this.state.value === TwakeServiceState.Stopped
) {
logger.info("Service %s is already stopped", this.name);
return this;
}
if (this.state.value !== TwakeServiceState.Started) {
logger.info("Service %s can not be stopped until started", this.name);
return this;
}
try {
logger.info("Stopping service %s", this.name);
this.state.next(TwakeServiceState.Stopping);
await this.doStop();
this.state.next(TwakeServiceState.Stopped);
logger.info("Service %s is stopped", this.name);
return this;
} catch (err) {
logger.error("Error while stopping service %s", this.name, err);
logger.error(err);
this.state.error(new Error(`Error while stopping service ${this.name}`));
throw err;
}
}
async doStop(): Promise<this> {
return this;
}
}
@@ -0,0 +1,21 @@
import { PREFIX_METADATA } from "./constants";
import { FastifyInstance, FastifyPluginCallback } from "fastify";
import gr from "../../../../services/global-resolver";
export abstract class AbstractWebService {
abstract routes(fastify: FastifyInstance): void;
public get prefix(): string {
return Reflect.getMetadata(PREFIX_METADATA, this) || "/";
}
public process(): void {
const wrapper: FastifyPluginCallback = (fastify: FastifyInstance, options, next) => {
this.routes(fastify);
next();
};
gr.fastify.register(wrapper, { prefix: this.prefix });
}
}
@@ -0,0 +1,32 @@
import { IConfig } from "config";
import configuration from "../../config";
import { TwakeServiceConfiguration } from "./api";
export class Configuration implements TwakeServiceConfiguration {
configuration: IConfig;
serviceConfiguration: IConfig;
constructor(path: string) {
try {
this.serviceConfiguration = configuration.get(path);
} catch {
// NOP
}
}
get<T>(name?: string, defaultValue?: T): T {
let value: T;
try {
value = this.serviceConfiguration as unknown as T;
if (name) {
value =
this.serviceConfiguration &&
(this.serviceConfiguration.get(name) || configuration.get(name));
}
} catch {
value = defaultValue || null;
} finally {
return value;
}
}
}
@@ -0,0 +1,8 @@
import { CONSUMES_METADATA } from "../api";
export function Consumes(services: string[] = []): ClassDecorator {
// eslint-disable-next-line @typescript-eslint/ban-types
return function (target: Function): void {
Reflect.defineMetadata(CONSUMES_METADATA, services, target.prototype);
};
}
@@ -0,0 +1,4 @@
export * from "./consumes";
export * from "./prefix";
export * from "./service-name";
export * from "./realtime";
@@ -0,0 +1,8 @@
import { PREFIX_METADATA } from "../api";
export function Prefix(prefix: string = "/"): ClassDecorator {
// eslint-disable-next-line @typescript-eslint/ban-types
return function (target: Function): void {
Reflect.defineMetadata(PREFIX_METADATA, prefix, target.prototype);
};
}
@@ -0,0 +1,41 @@
import { CreateResult } from "../../api/crud-service";
import { RealtimeEntityEvent, RealtimeEntityActionType } from "../../../services/realtime/types";
import { websocketEventBus } from "../../../services/realtime/bus";
import { getRealtimeRecipients, getRoom, RealtimeRecipients } from ".";
/**
*
* @param path the path to push the notification to
* @param resourcePath the path of the resource itself
*/
export function RealtimeCreated<T>(recipients: RealtimeRecipients<T>): MethodDecorator {
// eslint-disable-next-line @typescript-eslint/ban-types
return function (target: Object, propertyKey: string, descriptor: PropertyDescriptor): void {
const originalMethod = descriptor.value;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
descriptor.value = async function (...args: any[]) {
const result: CreateResult<T> = await originalMethod.apply(this, args);
// context should always be the last arg
const context = args && args[args.length - 1];
if (!(result instanceof CreateResult)) {
return result;
}
getRealtimeRecipients(recipients, result.entity, context).forEach(
({ room, path, resource }) => {
websocketEventBus.publish<T>(RealtimeEntityActionType.Created, {
type: result.type,
room: getRoom(room, result, context),
resourcePath: path || "/",
entity: resource,
result,
} as RealtimeEntityEvent<T>);
},
);
return result;
};
};
}
@@ -0,0 +1,41 @@
import { getRealtimeRecipients, getRoom, RealtimeRecipients } from "..";
import { DeleteResult } from "../../api/crud-service";
import { RealtimeEntityEvent, RealtimeEntityActionType } from "../../../services/realtime/types";
import { websocketEventBus } from "../../../services/realtime/bus";
/**
*
* @param path the path to push the notification to
* @param resourcePath the path of the resource itself
*/
export function RealtimeDeleted<T>(recipients: RealtimeRecipients<T>): MethodDecorator {
// eslint-disable-next-line @typescript-eslint/ban-types
return function (target: Object, propertyKey: string, descriptor: PropertyDescriptor): void {
const originalMethod = descriptor.value;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
descriptor.value = async function (...args: any[]) {
const result: DeleteResult<T> = await originalMethod.apply(this, args);
const context = args && args[args.length - 1];
if (!(result instanceof DeleteResult)) {
return result;
}
if (result.deleted)
getRealtimeRecipients(recipients, result.entity, context).forEach(
({ room, path, resource }) => {
websocketEventBus.publish<T>(RealtimeEntityActionType.Deleted, {
type: result.type,
room: getRoom(room, result, context),
resourcePath: path,
entity: resource,
result,
} as RealtimeEntityEvent<T>);
},
);
return result;
};
};
}
@@ -0,0 +1,67 @@
import { ResourcePath } from "../../../../../core/platform/services/realtime/types";
import { EntityTarget, ExecutionContext } from "../../api/crud-service";
export * from "./created";
export * from "./deleted";
export * from "./updated";
export * from "./saved";
export type RealtimeRecipients<T> =
| RealtimeRecipient<T>
| RealtimeRecipient<T>[]
| ((type: T, context?: ExecutionContext) => RealtimeRecipient<T>[] | RealtimeRecipient<T>);
export type RealtimeRecipient<T> = {
room: RealtimePath<T>;
path?: string;
resource?: any | T;
};
export function getRealtimeRecipients<T>(
recipients: RealtimeRecipients<T>,
type: T,
context?: ExecutionContext,
): RealtimeRecipient<T>[] {
if (typeof recipients === "function") recipients = recipients(type, context);
if (!recipients) return [];
if (!(recipients as RealtimeRecipient<T>[]).length)
recipients = [recipients as RealtimeRecipient<T>];
return (recipients as RealtimeRecipient<T>[]).map(recipient => {
return {
resource: recipient.resource || type,
path: recipient.path || "/",
room: recipient.room,
};
});
}
export type RealtimePath<T> = string | ResourcePath | ResourcePathResolver<T>;
export interface ResourcePathResolver<T> {
(type: T, context?: ExecutionContext): ResourcePath;
}
export interface PathResolver<T> {
(type: T, context?: ExecutionContext): string;
}
export type RealtimeEntity<T> = null | ((type: T, context?: ExecutionContext) => T);
export function getRoom<T>(
path: RealtimePath<T> = ResourcePath.default(),
entity: EntityTarget<T>,
context: ExecutionContext,
): ResourcePath {
if (typeof path === "string") {
return ResourcePath.get(path);
}
return typeof path === "function" ? path(entity.entity, context) : path;
}
export function getPath<T>(
path: string | PathResolver<T> = "/",
entity: EntityTarget<T>,
context: ExecutionContext,
): string {
return typeof path === "function" ? path(entity.entity, context) : path;
}
@@ -0,0 +1,41 @@
import { getRealtimeRecipients, getRoom, RealtimeRecipients } from "..";
import { SaveResult } from "../../api/crud-service";
import { RealtimeEntityEvent, RealtimeEntityActionType } from "../../../services/realtime/types";
import { websocketEventBus } from "../../../services/realtime/bus";
/**
*
* @param path the path to push the notification to
* @param resourcePath the path of the resource itself
*/
export function RealtimeSaved<T>(recipients: RealtimeRecipients<T>): MethodDecorator {
// eslint-disable-next-line @typescript-eslint/ban-types
return function (target: Object, propertyKey: string, descriptor: PropertyDescriptor): void {
const originalMethod = descriptor.value;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
descriptor.value = async function (...args: any[]) {
const result: SaveResult<T> = await originalMethod.apply(this, args);
// context should always be the last arg
const context = args && args[args.length - 1];
if (!(result instanceof SaveResult)) {
return result;
}
getRealtimeRecipients(recipients, result.entity, context).forEach(
({ room, path, resource }) => {
websocketEventBus.publish<T>(RealtimeEntityActionType.Saved, {
type: result.type,
room: getRoom(room, result, context),
resourcePath: path,
entity: resource,
result,
} as RealtimeEntityEvent<T>);
},
);
return result;
};
};
}
@@ -0,0 +1,40 @@
import { getRealtimeRecipients, getRoom, RealtimeRecipients } from "..";
import { UpdateResult } from "../../api/crud-service";
import { RealtimeEntityEvent, RealtimeEntityActionType } from "../../../services/realtime/types";
import { websocketEventBus } from "../../../services/realtime/bus";
/**
*
* @param path the path to push the notification to
* @param resourcePath the path of the resource itself
*/
export function RealtimeUpdated<T>(recipients: RealtimeRecipients<T>): MethodDecorator {
// eslint-disable-next-line @typescript-eslint/ban-types
return function (target: Object, propertyKey: string, descriptor: PropertyDescriptor): void {
const originalMethod = descriptor.value;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
descriptor.value = async function (...args: any[]) {
const result: UpdateResult<T> = await originalMethod.apply(this, args);
const context = args && args[args.length - 1];
if (!(result instanceof UpdateResult)) {
return result;
}
getRealtimeRecipients(recipients, result.entity, context).forEach(
({ room, path, resource }) => {
websocketEventBus.publish<T>(RealtimeEntityActionType.Updated, {
type: result.type,
room: getRoom(room, result, context),
resourcePath: path,
entity: resource,
result,
} as RealtimeEntityEvent<T>);
},
);
return result;
};
};
}
@@ -0,0 +1,10 @@
import { SERVICENAME_METADATA } from "../api";
export function ServiceName(name: string): ClassDecorator {
// eslint-disable-next-line @typescript-eslint/ban-types
return function (target: Function): void {
if (name) {
Reflect.defineMetadata(SERVICENAME_METADATA, name, target.prototype);
}
};
}
@@ -0,0 +1,35 @@
import { getLogger } from "../logger";
const logger = getLogger("core.platform.framework.decorators.Skip");
type SkipCondition = () => Promise<boolean> | boolean;
/**
* Skip a method call based on some condition
*/
export function Skip(condition: SkipCondition): MethodDecorator {
// eslint-disable-next-line @typescript-eslint/ban-types
return function (target: Object, _propertyKey: string, descriptor: PropertyDescriptor): void {
logger.trace("Skipping method call");
const originalMethod = descriptor.value;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
descriptor.value = async function (...args: any[]) {
const skipIt = await (condition && condition());
if (skipIt) {
return target;
}
return await originalMethod.apply(this, args);
};
};
}
/**
* Skip when process.env.NODE_ENV is set to "cli"
*/
export function SkipCLI(): MethodDecorator {
return Skip(() => process.env.NODE_ENV === "cli");
}
@@ -0,0 +1,8 @@
import { PREFIX_METADATA } from "../api";
export function WebService(prefix: string = "/"): ClassDecorator {
// eslint-disable-next-line @typescript-eslint/ban-types
return function (target: Function): void {
Reflect.defineMetadata(PREFIX_METADATA, prefix, target.prototype);
};
}
@@ -0,0 +1,47 @@
import { Subject } from "rxjs";
import { logger as rootLogger } from "./logger";
import { ExecutionContext } from "./api/crud-service";
const logger = rootLogger.child({
component: "twake.core.platform.framework.event-bus",
});
/**
* A local event bus in the platform. Used by platform components and services to communicate using publish subscribe.
*/
class EventBus {
private subjects: Map<string, Subject<any>>;
constructor() {
this.subjects = new Map<string, Subject<any>>();
}
subscribe<T>(name: string, listener: (_data: T) => void): this {
if (!this.subjects.has(name)) {
this.subjects.set(name, new Subject<T>());
}
this.subjects.get(name).subscribe({
next: (value: T) => {
try {
listener(value);
} catch (err) {
logger.warn({ err }, "Error while calling listener");
}
},
});
return this;
}
publish<T>(name: string, data: T, _context?: ExecutionContext): boolean {
if (!this.subjects.has(name)) {
return false;
}
this.subjects.get(name)?.next(data);
return true;
}
}
export const localEventBus = new EventBus();
@@ -0,0 +1,32 @@
import {
TwakeContext,
TwakeService,
TwakeServiceConfiguration,
TwakeServiceOptions,
TwakeServiceProvider,
} from "./api";
import { Configuration } from "./configuration";
class StaticTwakeServiceFactory {
public async create<
T extends TwakeService<TwakeServiceProvider> = TwakeService<TwakeServiceProvider>,
>(
module: { new (options?: TwakeServiceOptions<TwakeServiceConfiguration>): T },
context: TwakeContext,
configuration?: string,
): Promise<T> {
let config;
if (configuration) {
config = new Configuration(configuration);
}
const instance = new module({ configuration: config });
instance.context = context;
return instance;
}
}
export const TwakeServiceFactory = new StaticTwakeServiceFactory();
@@ -0,0 +1,8 @@
import "reflect-metadata";
export * from "./api";
export * from "./decorators";
export * from "./configuration";
export * from "./factory";
export * from "./logger";
export * from "./utils/component-utils";
@@ -0,0 +1,21 @@
import pino from "pino";
import { Configuration } from "./configuration";
const config = new Configuration("logger");
export type TwakeLogger = pino.Logger;
export const logger = pino({
name: "TwakeApp",
level: config.get("level", "info") || "info",
prettyPrint:
process.env.NODE_ENV?.indexOf("test") > -1
? {
translateTime: "HH:MM:ss Z",
ignore: "pid,hostname,name",
}
: false,
});
export const getLogger = (name?: string): TwakeLogger =>
logger.child({ name: `twake${name ? "." + name : ""}` });

Some files were not shown because too many files have changed in this diff Show More