Implement trackedEntities transfer

README.md

@@ -763,3 +763,17 @@ File option can be a file or directory path, if its a directory path the file wi
 CSV headers:
 
 dataElement ID | dataElement Name | categoryOptionCombo ID | categoryOptionCombo Name | Value
+
+## Tracked Entities
+
+### Transfer
+
+Transfer tracked entities to another org unit, using a CSV as source data (expected columns: trackedEntityId, newOrgUnitId):
+
+```console
+shell:~$ yarn start trackedEntities transfer \
+  --url=http://localhost:8080 \
+  --auth="USER:PASSWORD"  \
+  --input-file=transfers.csv \
+  --post
+```

src/data/TrackedEntityD2Repository.ts

@@ -1,16 +1,18 @@
 import _ from "lodash";
 import { D2Api } from "types/d2-api";
 import { Async } from "domain/entities/Async";
+import logger from "utils/log";
 
 import { getInChunks } from "./dhis2-utils";
 import { Stats } from "domain/entities/Stats";
 import {
     TrackedEntityFilterParams,
     TrackedEntityRepository,
 } from "domain/repositories/TrackedEntityRepository";
-import { TrackedEntity } from "domain/entities/TrackedEntity";
+import { TrackedEntity, TrackedEntityTransfer } from "domain/entities/TrackedEntity";
 import { D2TrackedEntity } from "./ProgramsD2Repository";
 import { D2Tracker } from "./D2Tracker";
+import { TrackedEntityInstance } from "@eyeseetea/d2-api/api/trackedEntityInstances";
 
 export class TrackedEntityD2Repository implements TrackedEntityRepository {
     private d2Tracker: D2Tracker;
@@ -54,7 +56,7 @@ export class TrackedEntityD2Repository implements TrackedEntityRepository {
             .keyBy(tei => tei.id)
             .value();
 
-        const stats = await getInChunks<Stats>(teisToFetch, async teiIds => {
+        const stats = await getInChunks<string, Stats>(teisToFetch, async teiIds => {
             const trackedEntities = await this.d2Tracker.getFromTracker<D2TrackedEntity>("trackedEntities", {
                 orgUnitIds: undefined,
                 programIds: programsIds,
@@ -94,4 +96,95 @@ export class TrackedEntityD2Repository implements TrackedEntityRepository {
 
         return Stats.combine(stats);
     }
+
+    async transfer(transfers: TrackedEntityTransfer[], options: { post: boolean }): Async<void> {
+        const teis = await this.getTeisFromTransfers(transfers);
+        const teisWithChanges = this.getTeisWithChanges(transfers, teis);
+        this.transferTeis(teisWithChanges, options);
+    }
+
+    private async transferTeis(teis: TrackedEntityInstance[], options: { post: boolean }) {
+        const { api } = this;
+
+        if (!options.post) {
+            logger.info(`Add --post to update tracked entities`);
+            return;
+        }
+
+        logger.info(`Transfer ownership: ${teis.length}`);
+
+        for (const tei of teis) {
+            await this.transferTei(tei);
+        }
+
+        await this.postTeis(teis, api);
+    }
+
+    private async transferTei(tei: TrackedEntityInstance) {
+        const programId = tei.enrollments[0]?.program;
+
+        if (!programId) {
+            logger.warn(`Tei without enrollments: ${tei.trackedEntityInstance}`);
+            return;
+        }
+
+        logger.debug(`Transfer ownership: trackedEntity.id=${tei.trackedEntityInstance}`);
+
+        const res = await this.api
+            .put<{ status: string }>(
+                `tracker/ownership/transfer?trackedEntityInstance=${tei.trackedEntityInstance}&ou=${tei.orgUnit}&program=${programId}`
+            )
+            .getData();
+
+        if (res.status !== "OK") {
+            logger.error(`Transfer ownership failed: ${JSON.stringify(res)}`);
+        }
+    }
+
+    private async postTeis(teis: TrackedEntityInstance[], api: D2Api) {
+        logger.info(`Import trackedEntities: ${teis.length}`);
+        const res = await api.trackedEntityInstances.post({}, { trackedEntityInstances: teis }).getData();
+        logger.info(`Tracked entities updated: ${res.status} - updated=${res.updated}`);
+    }
+
+    private getTeisWithChanges(transfers: TrackedEntityTransfer[], teis: TrackedEntityInstance[]) {
+        const rowsByTeiId = _.keyBy(transfers, row => row.trackedEntityId);
+
+        const teis2 = _(teis)
+            .sortBy(tei => tei.trackedEntityInstance)
+            .map(tei => {
+                const row = rowsByTeiId[tei.trackedEntityInstance];
+                if (!row) throw new Error("internal");
+                const orgUnitId = row.newOrgUnitId;
+                if (!orgUnitId) throw new Error("internal");
+                const hasChanges = tei.orgUnit !== orgUnitId;
+
+                return hasChanges ? { ...tei, orgUnit: orgUnitId } : undefined;
+            })
+            .compact()
+            .value();
+
+        logger.info(`trackedEntities that need to be transfered: ${teis2.length}`);
+        return teis2;
+    }
+
+    private async getTeisFromTransfers(transfers: TrackedEntityTransfer[]) {
+        logger.debug(`Get trackedEntities: ${transfers.length}`);
+
+        const teis = await getInChunks(transfers, async rowsChunk => {
+            const res = await this.api.trackedEntityInstances
+                .getAll({
+                    totalPages: true,
+                    ouMode: "ALL",
+                    trackedEntityInstance: rowsChunk.map(row => row.trackedEntityId).join(";"),
+                })
+                .getData();
+
+            return res.trackedEntityInstances;
+        });
+
+        logger.debug(`trackedEntities from server: ${teis.length}`);
+
+        return teis;
+    }
 }

src/domain/entities/TrackedEntity.ts

@@ -13,3 +13,8 @@ export type AttributeValue = {
     value: string;
     storedBy: string;
 };
+
+export type TrackedEntityTransfer = {
+    trackedEntityId: Id;
+    newOrgUnitId: Id;
+};

src/domain/repositories/TrackedEntityRepository.ts

@@ -1,11 +1,12 @@
 import { Async } from "domain/entities/Async";
 import { Id } from "domain/entities/Base";
 import { Stats } from "domain/entities/Stats";
-import { TrackedEntity } from "domain/entities/TrackedEntity";
+import { TrackedEntity, TrackedEntityTransfer } from "domain/entities/TrackedEntity";
 
 export interface TrackedEntityRepository {
     getAll(params: TrackedEntityFilterParams): Async<TrackedEntity[]>;
     save(trackedEntities: TrackedEntity[]): Async<Stats>;
+    transfer(trackedEntities: TrackedEntityTransfer[], options: { post: boolean }): Async<void>;
 }
 
 export type TrackedEntityFilterParams = {

src/domain/usecases/TransferTrackedEntitiesUseCase.ts

@@ -0,0 +1,11 @@
+import { Async } from "domain/entities/Async";
+import { TrackedEntityTransfer } from "domain/entities/TrackedEntity";
+import { TrackedEntityRepository } from "domain/repositories/TrackedEntityRepository";
+
+export class TransferTrackedEntitiesUseCase {
+    constructor(private trackedEntityRepository: TrackedEntityRepository) {}
+
+    execute(trackedEntityTransfers: TrackedEntityTransfer[], options: { post: boolean }): Async<void> {
+        return this.trackedEntityRepository.transfer(trackedEntityTransfers, options);
+    }
+}

src/scripts/cli.ts

@@ -14,6 +14,7 @@ import * as users from "./commands/users";
 import * as loadTesting from "./commands/loadTesting";
 import * as categoryoptions from "./commands/categoryoptions";
 import * as indicators from "./commands/indicators";
+import * as trackedEntities from "./commands/trackedEntities";
 
 export function runCli() {
     const cliSubcommands = subcommands({
@@ -26,6 +27,7 @@ export function runCli() {
             datavalues: dataValues.getCommand(),
             notifications: notifications.getCommand(),
             events: events.getCommand(),
+            trackedEntities: trackedEntities.getCommand(),
             sync: sync.getCommand(),
             users: users.getCommand(),
             usermonitoring: usermonitoring.getCommand(),

src/scripts/commands/trackedEntities.ts

@@ -0,0 +1,78 @@
+import { createReadStream } from "fs";
+import CsvReader from "csv-reader";
+import _ from "lodash";
+import { command, string, subcommands, option, flag } from "cmd-ts";
+import { getApiUrlOptions, getD2ApiFromArgs } from "scripts/common";
+import { Maybe } from "utils/ts-utils";
+import { TrackedEntityD2Repository } from "data/TrackedEntityD2Repository";
+import { TransferTrackedEntitiesUseCase } from "domain/usecases/TransferTrackedEntitiesUseCase";
+import { TrackedEntityTransfer } from "domain/entities/TrackedEntity";
+
+export function getCommand() {
+    return subcommands({
+        name: "trackedEntities",
+        cmds: {
+            transfer: transferTeisCommand,
+        },
+    });
+}
+
+const transferTeisCommand = command({
+    name: "transfer",
+    description: "Transfer tracked entities between organisation units",
+    args: {
+        ...getApiUrlOptions(),
+        inputCsvFile: option({
+            type: string,
+            long: "input-file",
+            description:
+                "CSV file to read tracked entities from (expected columns: trackedEntityId, newOrgUnitId)",
+        }),
+        post: flag({
+            long: "post",
+            description: "Execute transfer actions",
+        }),
+    },
+    handler: async args => {
+        const api = getD2ApiFromArgs(args);
+        const sourceRows = await readCSV<keyof TrackedEntityTransfer>(args.inputCsvFile);
+
+        const transfers = _(sourceRows)
+            .map((sourceRow): Maybe<TrackedEntityTransfer> => {
+                const { trackedEntityId, newOrgUnitId } = sourceRow;
+                return trackedEntityId && newOrgUnitId ? { trackedEntityId, newOrgUnitId } : undefined;
+            })
+            .compact()
+            .value();
+
+        const trackedEntityRepo = new TrackedEntityD2Repository(api);
+
+        await new TransferTrackedEntitiesUseCase(trackedEntityRepo).execute(transfers, args);
+    },
+});
+
+async function readCSV<Column extends string>(csvFilePath: string): Promise<Array<Record<Column, string>>> {
+    return new Promise((resolve, reject) => {
+        const inputStream = createReadStream(csvFilePath, "utf8");
+        const results: Array<Record<Column, string>> = [];
+
+        const reader = new CsvReader({
+            parseNumbers: false,
+            parseBooleans: false,
+            trim: true,
+            asObject: true,
+        });
+
+        inputStream
+            .pipe(reader)
+            .on("data", row => {
+                results.push(row as Record<Column, string>);
+            })
+            .on("end", () => {
+                resolve(results);
+            })
+            .on("error", error => {
+                reject(error);
+            });
+    });
+}