feat(core-manager): implement events listener and events database

__tests__/unit/core-manager/action-reader.test.ts

@@ -13,6 +13,7 @@ beforeEach(() => {
 
     sandbox.app.bind(Identifiers.ActionReader).to(ActionReader).inSingletonScope();
     sandbox.app.bind(Identifiers.SnapshotsManager).toConstantValue({});
+    sandbox.app.bind(Identifiers.WatcherDatabaseService).toConstantValue({});
     sandbox.app.bind(Container.Identifiers.PluginConfiguration).toConstantValue({});
     sandbox.app.bind(Container.Identifiers.FilesystemService).toConstantValue({});
 

__tests__/unit/core-manager/database-service.test.ts

@@ -0,0 +1,135 @@
+import "jest-extended";
+
+import { Container } from "@arkecosystem/core-kernel";
+import { Sandbox } from "@packages/core-test-framework";
+import { DatabaseService } from "@arkecosystem/core-manager/src/database-service";
+import { existsSync } from "fs-extra";
+import { dirSync, setGracefulCleanup } from "tmp";
+
+let sandbox: Sandbox;
+let database: DatabaseService;
+let storagePath: string;
+
+beforeEach(() => {
+    sandbox = new Sandbox();
+
+    storagePath = dirSync().name + "/events.sqlite";
+
+    sandbox.app.bind(Container.Identifiers.WatcherDatabaseService).to(DatabaseService).inSingletonScope();
+
+    sandbox.app.bind(Container.Identifiers.PluginConfiguration).toConstantValue({
+        getRequired: jest.fn().mockReturnValue(storagePath),
+    });
+
+    database = sandbox.app.get(Container.Identifiers.WatcherDatabaseService);
+});
+
+afterEach(() => {
+    setGracefulCleanup();
+});
+
+describe("DatabaseService", () => {
+    describe("Boot", () => {
+        it("should boot and create file", async () => {
+            database.boot();
+
+            expect(existsSync(storagePath)).toBeTrue();
+        });
+    });
+
+    describe("Dispose", () => {
+        it("should dispose", async () => {
+            database.boot();
+            database.dispose();
+
+            expect(() => {
+                database.addEvent("dummy_event", { data: "dummy_data" });
+            }).toThrowError();
+        });
+    });
+
+    describe("Flush", () => {
+        it("should dispose", async () => {
+            database.boot();
+
+            database.addEvent("dummy_event", { data: "dummy_data" });
+
+            expect(database.getAllEvents().length).toBe(1);
+
+            database.flush();
+
+            expect(database.getAllEvents().length).toBe(0);
+        });
+    });
+
+    describe("AddEvent", () => {
+        it("should add event", async () => {
+            database.boot();
+            expect(existsSync(storagePath)).toBeTrue();
+
+            database.addEvent("dummy_event", { data: "dummy_data" });
+
+            const result = database.getAllEvents();
+
+            expect(result).toBeArray();
+            expect(result[0]).toMatchObject({
+                id: 1,
+                event: "dummy_event",
+                data: { data: "dummy_data" },
+            });
+
+            expect(result[0].timestamp).toBeDefined();
+        });
+    });
+
+    describe("Query", () => {
+        beforeEach(() => {
+            database.boot();
+
+            for (let i = 0; i < 100; i++) {
+                database.addEvent("dummy_event", { data: "dummy_data" });
+                database.addEvent("another_dummy_event", { data: "another_dummy_data" });
+            }
+        });
+
+        it("should return limit 10", async () => {
+            const result = database.queryEvents();
+
+            expect(result.total).toBe(200);
+            expect(result.limit).toBe(10);
+            expect(result.offset).toBe(0);
+            expect(result.data).toBeArray();
+            expect(result.data.length).toBe(10);
+        });
+
+        it("should return limit 10 with offset", async () => {
+            const result = database.queryEvents({ offset: 10 });
+
+            expect(result.total).toBe(200);
+            expect(result.limit).toBe(10);
+            expect(result.offset).toBe(10);
+            expect(result.data).toBeArray();
+            expect(result.data.length).toBe(10);
+        });
+
+        it("should return limit 20", async () => {
+            const result = database.queryEvents({ limit: 20 });
+
+            expect(result.total).toBe(200);
+            expect(result.limit).toBe(20);
+            expect(result.offset).toBe(0);
+            expect(result.data).toBeArray();
+            expect(result.data.length).toBe(20);
+        });
+
+        it("should return events with name", async () => {
+            const result = database.queryEvents({ limit: 1000, event: "dummy_event" });
+
+            expect(result.total).toBe(100);
+            expect(result.limit).toBe(1000);
+            expect(result.offset).toBe(0);
+            expect(result.data).toBeArray();
+            expect(result.data.length).toBe(100);
+        });
+    });
+});

__tests__/unit/core-manager/listener.test.ts

@@ -0,0 +1,43 @@
+import "jest-extended";
+
+import { Container } from "@arkecosystem/core-kernel";
+import { Listener } from "@arkecosystem/core-manager/src/listener";
+import { Sandbox } from "@packages/core-test-framework";
+
+let sandbox: Sandbox;
+let listener: Listener;
+const mockDatabase = {
+    addEvent: jest.fn(),
+};
+
+let handler;
+
+const mockEventDispatcher = {
+    listen: jest.fn().mockImplementation((event, cb) => {
+        handler = cb;
+    }),
+};
+
+beforeEach(() => {
+    sandbox = new Sandbox();
+
+    sandbox.app.bind(Container.Identifiers.WatcherDatabaseService).toConstantValue(mockDatabase);
+    sandbox.app.bind(Container.Identifiers.EventDispatcherService).toConstantValue(mockEventDispatcher);
+    sandbox.app.bind(Container.Identifiers.WatcherEventListener).to(Listener).inSingletonScope();
+
+    listener = sandbox.app.get(Container.Identifiers.WatcherEventListener);
+});
+
+afterEach(() => {});
+
+describe("Listener", () => {
+    describe("Boot", () => {
+        it("should boot and save event on emit", async () => {
+            listener.boot();
+
+            handler.handle({ name: "dummy_event", data: "dummy_data" });
+
+            expect(mockDatabase.addEvent).toHaveBeenCalledTimes(1);
+        });
+    });
+});

__tests__/unit/core-manager/service-provider.test.ts

@@ -5,6 +5,7 @@ import { defaults } from "@packages/core-manager/src/defaults";
 import { Identifiers } from "@packages/core-manager/src/ioc";
 import { ServiceProvider } from "@packages/core-manager/src/service-provider";
 import path from "path";
+import { dirSync, setGracefulCleanup } from "tmp";
 
 let app: Application;
 
@@ -14,6 +15,10 @@ const logger = {
     error: jest.fn(),
 };
 
+const mockEventDispatcher = {
+    listen: jest.fn(),
+};
+
 const setPluginConfiguration = (app: Application, serviceProvider: ServiceProvider, configuration: any) => {
     const pluginConfiguration = app.get<Providers.PluginConfiguration>(Container.Identifiers.PluginConfiguration);
     const instance: Providers.PluginConfiguration = pluginConfiguration.from("core-monitor", configuration);
@@ -27,11 +32,17 @@ beforeEach(() => {
     app.bind(Container.Identifiers.LogService).toConstantValue(logger);
     app.bind(Container.Identifiers.PluginConfiguration).to(Providers.PluginConfiguration).inSingletonScope();
     app.bind(Container.Identifiers.FilesystemService).toConstantValue({});
+    app.bind(Container.Identifiers.EventDispatcherService).toConstantValue(mockEventDispatcher);
 
+    defaults.storage = dirSync().name + "/events.sqlite";
     defaults.server.https.tls.key = path.resolve(__dirname, "./__fixtures__/key.pem");
     defaults.server.https.tls.cert = path.resolve(__dirname, "./__fixtures__/server.crt");
 });
 
+afterEach(() => {
+    setGracefulCleanup();
+});
+
 describe("ServiceProvider", () => {
     let serviceProvider: ServiceProvider;
 

__tests__/unit/core-snapshots/service-provider.test.ts

@@ -1,20 +1,21 @@
 import "jest-extended";
-import * as typeorm from "typeorm";
+
 import { Container } from "@packages/core-kernel";
 import { ServiceProvider } from "@packages/core-snapshots/src";
 import { Sandbox } from "@packages/core-test-framework";
+import * as typeorm from "typeorm";
 
 let sandbox: Sandbox;
 
-let spyOnGetCustomRepository = jest.spyOn(typeorm, "getCustomRepository").mockReturnValue(undefined);
-let spyOnCreateConnection = jest.spyOn(typeorm, "createConnection").mockResolvedValue({
+const spyOnGetCustomRepository = jest.spyOn(typeorm, "getCustomRepository").mockReturnValue(undefined);
+const spyOnCreateConnection = jest.spyOn(typeorm, "createConnection").mockResolvedValue({
     close: jest.fn(),
 } as any);
 
 ServiceProvider.prototype.config = jest.fn().mockReturnValue({
-    all() {
-        return {};
-    },
+    all: jest.fn().mockReturnValue({
+        connection: {},
+    }),
 });
 
 beforeEach(() => {

packages/core-kernel/src/ioc/identifiers.ts

@@ -114,4 +114,8 @@ export const Identifiers = {
     // Registries
     TransactionHandlerRegistry: Symbol.for("Registry<TransactionHandler>"),
     TransactionHandlerProvider: Symbol.for("Provider<TransactionHandler>"),
+
+    // Watcher
+    WatcherEventListener: Symbol.for("Watcher<EventListener>"),
+    WatcherDatabaseService: Symbol.for("Watcher<DatabaseService>"),
 };

packages/core-manager/package.json

@@ -21,22 +21,21 @@
         "pretest": "bash ../../scripts/pre-test.sh"
     },
     "dependencies": {
-        "argon2": "^0.26.2",
         "@arkecosystem/core-cli": "^3.0.0-next.0",
-        "@arkecosystem/core-kernel": "^3.0.0-next.0",
         "@arkecosystem/core-database": "^3.0.0-next.0",
-        "@arkecosystem/core-cli": "^3.0.0-next.0",
+        "@arkecosystem/core-kernel": "^3.0.0-next.0",
         "@hapi/basic": "^6.0.0",
         "@hapi/boom": "^9.0.0",
         "@hapi/hapi": "^19.0.0",
         "@hapist/json-rpc": "^0.2.0",
         "@hapist/whitelist": "^0.1.0",
         "@sindresorhus/df": "^3.1.1",
+        "argon2": "^0.26.2",
+        "better-sqlite3": "^7.0.0",
         "got": "^11.1.3",
         "hapi-auth-bearer-token": "^6.1.6",
         "latest-version": "^5.1.0",
-        "require-from-string": "^2.0.2",
-        "typeorm": "^0.2.21"
+        "require-from-string": "^2.0.2"
     },
     "devDependencies": {
         "nock": "^12.0.3"

packages/core-manager/src/database-service.ts

@@ -0,0 +1,106 @@
+import { Container, Providers } from "@arkecosystem/core-kernel";
+import BetterSqlite3 from "better-sqlite3";
+import { ensureFileSync } from "fs-extra";
+
+@Container.injectable()
+export class DatabaseService {
+    @Container.inject(Container.Identifiers.PluginConfiguration)
+    @Container.tagged("plugin", "@arkecosystem/core-manager")
+    private readonly configuration!: Providers.PluginConfiguration;
+
+    private database!: BetterSqlite3.Database;
+
+    public boot(): void {
+        const filename = this.configuration.getRequired<string>("storage");
+        ensureFileSync(filename);
+
+        this.database = new BetterSqlite3(filename);
+        this.database.exec(`
+            PRAGMA journal_mode = WAL;
+            CREATE TABLE IF NOT EXISTS events (id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, event VARCHAR(255) NOT NULL, data JSON NOT NULL, timestamp DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP);
+        `);
+    }
+
+    public dispose(): void {
+        this.database.close();
+    }
+
+    public flush(): void {
+        this.database.prepare("DELETE FROM events").run();
+    }
+
+    public addEvent(event: string, data: any): void {
+        this.database.prepare("INSERT INTO events (event, data) VALUES (:event, json(:data))").run({
+            event: event,
+            data: JSON.stringify(data || {}),
+        });
+    }
+
+    public getAllEvents(): any[] {
+        return this.database
+            .prepare("SELECT * FROM events")
+            .pluck(false)
+            .all()
+            .map((x) => {
+                x.data = JSON.parse(x.data);
+                return x;
+            });
+    }
+
+    public getTotal(conditions?: any): number {
+        return this.database.prepare(`SELECT COUNT(*) FROM events ${this.prepareWhere(conditions)}`).get()[
+            "COUNT(*)"
+        ] as number;
+    }
+
+    public queryEvents(conditions?: any): any {
+        const limit = this.prepareLimit(conditions);
+        const offset = this.prepareOffset(conditions);
+
+        return {
+            total: this.getTotal(conditions),
+            limit,
+            offset,
+            data: this.database
+                .prepare(`SELECT * FROM events ${this.prepareWhere(conditions)} LIMIT ${limit} OFFSET ${offset}`)
+                .pluck(false)
+                .all()
+                .map((x) => {
+                    x.data = JSON.parse(x.data);
+                    return x;
+                }),
+        };
+    }
+
+    private prepareLimit(conditions?: any): number {
+        if (conditions?.limit && typeof conditions.limit === "number" && conditions.limit <= 1000) {
+            return conditions.limit;
+        }
+
+        return 10;
+    }
+
+    private prepareOffset(conditions?: any): number {
+        if (conditions?.offset && typeof conditions.offset === "number") {
+            return conditions.offset;
+        }
+
+        return 0;
+    }
+
+    private prepareWhere(conditions?: any): string {
+        let query = "";
+
+        if (!conditions) {
+            return query;
+        }
+
+        for (const key of Object.keys(conditions)) {
+            if (key === "event") {
+                query += `WHERE event LIKE '${conditions[key]}%'`;
+            }
+        }
+
+        return query;
+    }
+}