Load a whole organisation from a file, or none of it
Second half of the mass import: the transactional loader, the /import page
and a template generated from the same schema the validation uses.
Everything happens in one transaction. A half-loaded organisation — areas
without departments, positions without people — is worse than none, because
it looks like data. The dry run is the same code path with a rollback at the
end, so the report is built against the real current state rather than a
copy, and nothing is cached between checking and committing: the file is
sent twice. That costs one upload and avoids server-side state that can
expire, fill up, or be confused between two people.
Personnel numbers are taken from the file, not reassigned. personnel_number
is GENERATED ALWAYS AS IDENTITY, so this needs OVERRIDING SYSTEM VALUE and a
hand-written insert — worth it, because the number is on payslips, in files
and on badges. An import that reissues it is not a migration. The identity
counter is advanced afterwards; without that the next hire draws a number
the import already used, and the unique index refuses it weeks later, far
from the cause.
Three defects the first real run against the database exposed, none of which
typecheck, lint or 231 tests could have found:
- weekly_hours is bound to employment type by a CHECK constraint: full time
is exactly 38.5. The import reached the insert and was rolled back. Now
it is a finding with a row number.
- Titles are restricted to a fixed list by another CHECK. Same treatment.
- setval() needs UPDATE on the sequence, which `usage, select` does not
grant. Migration 20260803120000 adds it; until it is applied, an import
containing people will fail at the last step and take itself back.
I also had exit_date > entry_date where the database has >=. Someone who
never starts enters and leaves the same day; the stricter rule would have
rejected a real case.
Verified against the live database through the actual route and session: a
file with four deliberate faults produced exactly four findings, each with
sheet, row and column, and the rollback left nothing behind.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
335
lib/import/load.ts
Normal file
335
lib/import/load.ts
Normal file
@@ -0,0 +1,335 @@
|
||||
import "server-only";
|
||||
import { sql, type Tx } from "@/lib/db";
|
||||
import { todayIso } from "@/lib/format";
|
||||
import { deriveStatusAsOf } from "@/lib/reports";
|
||||
import { normalizeSvnr } from "@/lib/svnr";
|
||||
import type { Bestand, Datensatz, Zeile } from "./validate";
|
||||
|
||||
// Schreiben einer geprüften Datei.
|
||||
//
|
||||
// Der Aufruf steckt in **einer** Transaktion (siehe die Route). Das ist keine
|
||||
// Vorsicht, sondern die Bedingung: eine halb geladene Organisation — Bereiche
|
||||
// ohne Abteilungen, Planstellen ohne Personen — ist schlimmer als gar keine,
|
||||
// weil sie nach Daten aussieht. Bricht irgendetwas ab, war nichts.
|
||||
//
|
||||
// Angelegt wird nur; aktualisiert wird nie. Was es schon gibt, hat die
|
||||
// Prüfung vorher abgewiesen. Damit kann ein Tippfehler in einer Datei keine
|
||||
// bestehenden Personaldaten überschreiben.
|
||||
|
||||
/** Wie viele Zeilen je INSERT. Postgres verträgt 65535 Parameter je Anweisung. */
|
||||
const PORTION = 200;
|
||||
|
||||
export type Ladebericht = Record<string, number>;
|
||||
|
||||
const txt = (v: unknown): string | null => (typeof v === "string" && v !== "" ? v : null);
|
||||
const zahl = (v: unknown): number | null => (typeof v === "number" ? v : null);
|
||||
const bool = (v: unknown, ersatz: boolean): boolean => (typeof v === "boolean" ? v : ersatz);
|
||||
const liste = (v: unknown): string[] | null => (Array.isArray(v) ? (v as string[]) : null);
|
||||
|
||||
async function einfuegen<T>(tx: Tx, tabelle: string, zeilen: T[]): Promise<void> {
|
||||
for (let i = 0; i < zeilen.length; i += PORTION) {
|
||||
await tx
|
||||
// Die Tabellennamen stammen aus einer geschlossenen Aufzählung in
|
||||
// diesem Modul, nie aus der Datei.
|
||||
.insertInto(tabelle as never)
|
||||
.values(zeilen.slice(i, i + PORTION) as never)
|
||||
.execute();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Was bereits im System steht.
|
||||
*
|
||||
* Läuft in derselben Transaktion wie das Schreiben. Zwischen Prüfung und
|
||||
* Schreiben könnte sonst jemand dieselbe Planstelle besetzen, und der Import
|
||||
* liefe in den Teilindex statt in eine verständliche Meldung.
|
||||
*/
|
||||
export async function bestandLaden(tx: Tx): Promise<Bestand> {
|
||||
const heute = todayIso();
|
||||
|
||||
const [standorte, einheiten, jobs, stellen, besetzungen, personen] = await Promise.all([
|
||||
tx.selectFrom("locations").select(["id", "name"]).execute(),
|
||||
tx.selectFrom("org_units").select(["id", "org_number"]).execute(),
|
||||
tx.selectFrom("jobs").select(["id", "code"]).execute(),
|
||||
tx.selectFrom("om_positions").select(["id", "position_number"]).execute(),
|
||||
tx
|
||||
.selectFrom("position_assignments")
|
||||
.select(["position_id"])
|
||||
.where((eb) => eb.or([eb("valid_to", "is", null), eb("valid_to", ">=", heute)]))
|
||||
.execute(),
|
||||
tx.selectFrom("employees").select(["id", "personnel_number", "email", "sv_nummer"]).execute(),
|
||||
]);
|
||||
|
||||
const besetzt = new Set(besetzungen.map((b) => b.position_id));
|
||||
|
||||
return {
|
||||
standorte: new Map(standorte.map((l) => [l.name, l.id])),
|
||||
orgNummern: new Map(einheiten.map((o) => [o.org_number, o.id])),
|
||||
jobCodes: new Map(jobs.map((j) => [j.code, j.id])),
|
||||
planstellen: new Map(stellen.map((p) => [p.position_number, { id: p.id, besetzt: besetzt.has(p.id) }])),
|
||||
personalnummern: new Map(personen.map((e) => [e.personnel_number, e.id])),
|
||||
emails: new Set(personen.map((e) => e.email.toLowerCase())),
|
||||
svNummern: new Set(personen.filter((e) => e.sv_nummer).map((e) => normalizeSvnr(e.sv_nummer!))),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Ordnet Einheiten so, dass jede nach ihrer übergeordneten kommt.
|
||||
*
|
||||
* Der Fremdschlüssel auf parent_id wird je Zeile geprüft, und eine Datei
|
||||
* darf ihre Zeilen in beliebiger Reihenfolge führen — eine Mappe, die nach
|
||||
* Bezeichnung sortiert ist, hätte sonst Pech.
|
||||
*/
|
||||
function elternZuerst(zeilen: Zeile[], bekannt: Set<string>): { sortiert: Zeile[]; ungeloest: Zeile[] } {
|
||||
const offen = [...zeilen];
|
||||
const sortiert: Zeile[] = [];
|
||||
const erledigt = new Set(bekannt);
|
||||
|
||||
let fortschritt = true;
|
||||
while (offen.length > 0 && fortschritt) {
|
||||
fortschritt = false;
|
||||
for (let i = offen.length - 1; i >= 0; i--) {
|
||||
const eltern = txt(offen[i].werte.parent_org_number);
|
||||
if (!eltern || erledigt.has(eltern)) {
|
||||
const nummer = txt(offen[i].werte.org_number);
|
||||
if (nummer) erledigt.add(nummer);
|
||||
sortiert.push(offen[i]);
|
||||
offen.splice(i, 1);
|
||||
fortschritt = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
// Was übrig bleibt, hängt in einem Kreis — die Prüfung fängt den einfachen
|
||||
// Fall (sich selbst übergeordnet), längere Ketten fallen hier auf.
|
||||
return { sortiert, ungeloest: offen };
|
||||
}
|
||||
|
||||
export async function laden(
|
||||
tx: Tx,
|
||||
datensatz: Datensatz,
|
||||
bestand: Bestand,
|
||||
akteur: { userId: string; name: string }
|
||||
): Promise<Ladebericht> {
|
||||
const hole = (name: string) => datensatz[name] ?? [];
|
||||
const bericht: Ladebericht = {};
|
||||
const heute = todayIso();
|
||||
|
||||
// ── Standorte ─────────────────────────────────────────────────
|
||||
const standortId = new Map(bestand.standorte);
|
||||
const neueStandorte = hole("Standorte").map((z) => ({
|
||||
name: txt(z.werte.name)!,
|
||||
country: txt(z.werte.country)!,
|
||||
}));
|
||||
if (neueStandorte.length) {
|
||||
const zurueck = await tx.insertInto("locations").values(neueStandorte).returning(["id", "name"]).execute();
|
||||
for (const r of zurueck) standortId.set(r.name, r.id);
|
||||
}
|
||||
bericht.Standorte = neueStandorte.length;
|
||||
|
||||
// ── Organisation ──────────────────────────────────────────────
|
||||
const orgId = new Map(bestand.orgNummern);
|
||||
const { sortiert, ungeloest } = elternZuerst(hole("Organisation"), new Set(orgId.keys()));
|
||||
if (ungeloest.length) {
|
||||
throw new Error(
|
||||
`Die übergeordneten Einheiten von ${ungeloest.length} Zeile(n) lassen sich nicht auflösen — vermutlich ein Kreis in der Spalte „Übergeordnet“.`
|
||||
);
|
||||
}
|
||||
for (const z of sortiert) {
|
||||
const eltern = txt(z.werte.parent_org_number);
|
||||
const r = await tx
|
||||
.insertInto("org_units")
|
||||
.values({
|
||||
org_number: txt(z.werte.org_number)!,
|
||||
name: txt(z.werte.name)!,
|
||||
unit_type: txt(z.werte.unit_type) as never,
|
||||
parent_id: eltern ? orgId.get(eltern)! : null,
|
||||
valid_from: txt(z.werte.valid_from) ?? heute,
|
||||
valid_to: txt(z.werte.valid_to),
|
||||
})
|
||||
.returning(["id", "org_number"])
|
||||
.execute();
|
||||
orgId.set(r[0].org_number, r[0].id);
|
||||
}
|
||||
bericht.Organisation = sortiert.length;
|
||||
|
||||
// ── Jobkatalog ────────────────────────────────────────────────
|
||||
const jobId = new Map(bestand.jobCodes);
|
||||
const neueJobs = hole("Jobkatalog").map((z) => ({ code: txt(z.werte.code)!, title: txt(z.werte.title)! }));
|
||||
if (neueJobs.length) {
|
||||
const zurueck = await tx.insertInto("jobs").values(neueJobs).returning(["id", "code"]).execute();
|
||||
for (const r of zurueck) jobId.set(r.code, r.id);
|
||||
}
|
||||
bericht.Jobkatalog = neueJobs.length;
|
||||
|
||||
// ── Planstellen ───────────────────────────────────────────────
|
||||
const stellenId = new Map([...bestand.planstellen].map(([nr, p]) => [nr, p.id]));
|
||||
const neueStellen = hole("Planstellen").map((z) => ({
|
||||
position_number: txt(z.werte.position_number)!,
|
||||
org_unit_id: orgId.get(txt(z.werte.org_number)!)!,
|
||||
job_id: jobId.get(txt(z.werte.job_code)!)!,
|
||||
is_chief: bool(z.werte.is_chief, false),
|
||||
valid_from: txt(z.werte.valid_from) ?? heute,
|
||||
valid_to: txt(z.werte.valid_to),
|
||||
}));
|
||||
if (neueStellen.length) {
|
||||
const zurueck = await tx
|
||||
.insertInto("om_positions")
|
||||
.values(neueStellen)
|
||||
.returning(["id", "position_number"])
|
||||
.execute();
|
||||
for (const r of zurueck) stellenId.set(r.position_number, r.id);
|
||||
}
|
||||
bericht.Planstellen = neueStellen.length;
|
||||
|
||||
// ── Personen ──────────────────────────────────────────────────
|
||||
const personId = new Map(bestand.personalnummern);
|
||||
const personenZeilen = hole("Personen");
|
||||
const besetzungen: { position_id: string; employee_id: string; valid_from: string; valid_to: string | null }[] = [];
|
||||
|
||||
for (const z of personenZeilen) {
|
||||
const w = z.werte;
|
||||
const eintritt = txt(w.entry_date)!;
|
||||
const austritt = txt(w.exit_date);
|
||||
const karenzVon = txt(w.karenz_start_date);
|
||||
const karenzBis = txt(w.karenz_return_date);
|
||||
|
||||
const werte = {
|
||||
personnel_number: zahl(w.personnel_number)!,
|
||||
first_name: txt(w.first_name)!,
|
||||
last_name: txt(w.last_name)!,
|
||||
gender: txt(w.gender) as never,
|
||||
birth_date: txt(w.birth_date)!,
|
||||
sv_nummer: txt(w.sv_nummer) ? normalizeSvnr(txt(w.sv_nummer)!) : null,
|
||||
nationality: txt(w.nationality) ?? "Österreich",
|
||||
address: txt(w.address),
|
||||
postal_code: txt(w.postal_code),
|
||||
city: txt(w.city),
|
||||
address_country: txt(w.address_country),
|
||||
email: txt(w.email)!,
|
||||
phone: txt(w.phone),
|
||||
title_prefix: liste(w.title_prefix) ?? [],
|
||||
title_suffix: liste(w.title_suffix) ?? [],
|
||||
job_title: txt(w.job_title)!,
|
||||
location_id: standortId.get(txt(w.location)!)!,
|
||||
employment_type: (txt(w.employment_type) ?? "Vollzeit") as never,
|
||||
weekly_hours: zahl(w.weekly_hours) ?? 38.5,
|
||||
// Die Prüfung hat jeden Eintrag gegen die Wochentage abgeglichen und
|
||||
// auf die Schreibweise der Datenbank gebracht; hier steht deshalb
|
||||
// sicher nur Mo…So.
|
||||
work_days: (liste(w.work_days) ?? ["Mo", "Di", "Mi", "Do", "Fr"]) as never,
|
||||
contract_type: (txt(w.contract_type) ?? "unbefristet") as never,
|
||||
contract_end_date: txt(w.contract_end_date),
|
||||
paygrade: (txt(w.paygrade) ?? "B") as never,
|
||||
collective_agreement: (txt(w.collective_agreement) ?? "Handel") as never,
|
||||
worker_type: (txt(w.worker_type) ?? "Angestellte:r") as never,
|
||||
monthly_salary_gross: zahl(w.monthly_salary_gross),
|
||||
source: (txt(w.source) ?? "Extern") as never,
|
||||
is_betriebsrat: bool(w.is_betriebsrat, false),
|
||||
has_dienstwagen: bool(w.has_dienstwagen, false),
|
||||
is_laterale_fuehrung: bool(w.is_laterale_fuehrung, false),
|
||||
is_c_level: bool(w.is_c_level, false),
|
||||
entry_date: eintritt,
|
||||
exit_date: austritt,
|
||||
exit_reason: txt(w.exit_reason),
|
||||
karenz_start_date: karenzVon,
|
||||
karenz_return_date: karenzBis,
|
||||
absence_type: txt(w.absence_type),
|
||||
// Der Status wird **abgeleitet**, nicht importiert. Stünde er in der
|
||||
// Datei, könnte er den Daten widersprechen — jemand mit Austritt und
|
||||
// Status „Aktiv“ —, und die Anwendung leitet ihn ohnehin überall aus
|
||||
// denselben Datumsangaben ab.
|
||||
status: deriveStatusAsOf(
|
||||
{ entry_date: eintritt, exit_date: austritt, karenz_start_date: karenzVon, karenz_return_date: karenzBis },
|
||||
heute
|
||||
),
|
||||
};
|
||||
|
||||
// Von Hand geschrieben statt über den Abfragebauer, wegen genau eines
|
||||
// Wortes: OVERRIDING SYSTEM VALUE.
|
||||
//
|
||||
// personnel_number ist GENERATED ALWAYS AS IDENTITY — die Datenbank
|
||||
// vergibt sie und weist einen eigenen Wert sonst ab. Für eine Übernahme
|
||||
// aus einem Altsystem ist das die falsche Richtung: die Nummer steht auf
|
||||
// Lohnzetteln, in Akten und auf Ausweisen. Ein Import, der sie neu
|
||||
// würfelt, ist keine Übernahme.
|
||||
const spalten = Object.keys(werte);
|
||||
const r = await sql<{ id: string; personnel_number: number }>`
|
||||
insert into employees (${sql.raw(spalten.map((s) => `"${s}"`).join(", "))})
|
||||
overriding system value
|
||||
values (${sql.join(Object.values(werte).map((v) => sql.val(v)))})
|
||||
returning id, personnel_number
|
||||
`.execute(tx);
|
||||
|
||||
personId.set(r.rows[0].personnel_number, r.rows[0].id);
|
||||
|
||||
const stelle = txt(w.position_number);
|
||||
if (stelle) {
|
||||
besetzungen.push({
|
||||
position_id: stellenId.get(stelle)!,
|
||||
employee_id: r.rows[0].id,
|
||||
valid_from: eintritt,
|
||||
// Beim Austritt endet die Besetzung — sonst gälte die Planstelle als
|
||||
// belegt und liesse sich nicht nachbesetzen.
|
||||
valid_to: austritt,
|
||||
});
|
||||
}
|
||||
}
|
||||
bericht.Personen = personenZeilen.length;
|
||||
|
||||
// Den Zähler nachziehen. Ohne das vergibt die Datenbank für die nächste
|
||||
// Neueinstellung eine Nummer, die der Import bereits verbraucht hat — und
|
||||
// der eindeutige Index weist sie ab. Der Fehler träte erst Wochen später
|
||||
// auf, beim ersten Eintritt nach der Übernahme.
|
||||
if (personenZeilen.length > 0) {
|
||||
await sql`
|
||||
select setval(
|
||||
pg_get_serial_sequence('employees', 'personnel_number'),
|
||||
(select max(personnel_number) from employees)
|
||||
)
|
||||
`.execute(tx);
|
||||
}
|
||||
|
||||
await einfuegen(tx, "position_assignments", besetzungen);
|
||||
|
||||
// ── Historie und Angehörige ───────────────────────────────────
|
||||
const historie = hole("Historie").map((z) => ({
|
||||
employee_id: personId.get(zahl(z.werte.personnel_number)!)!,
|
||||
event_date: txt(z.werte.event_date)!,
|
||||
event_type: txt(z.werte.event_type)! as never,
|
||||
description: txt(z.werte.description)!,
|
||||
}));
|
||||
await einfuegen(tx, "employee_history", historie);
|
||||
bericht.Historie = historie.length;
|
||||
|
||||
const angehoerige = hole("Angehörige").map((z) => ({
|
||||
employee_id: personId.get(zahl(z.werte.personnel_number)!)!,
|
||||
first_name: txt(z.werte.first_name)!,
|
||||
last_name: txt(z.werte.last_name)!,
|
||||
relationship: txt(z.werte.relationship)!,
|
||||
birth_date: txt(z.werte.birth_date)!,
|
||||
sv_nummer: txt(z.werte.sv_nummer) ? normalizeSvnr(txt(z.werte.sv_nummer)!) : null,
|
||||
}));
|
||||
await einfuegen(tx, "employee_dependents", angehoerige);
|
||||
bericht.Angehörige = angehoerige.length;
|
||||
|
||||
// ── Protokoll ─────────────────────────────────────────────────
|
||||
// Ein Eintrag für den ganzen Vorgang, nicht einer je Zeile: 800 Zeilen
|
||||
// würden das Protokoll unlesbar machen, und der Vorgang ist ohnehin
|
||||
// untrennbar — er ist eine Transaktion.
|
||||
const zusammenfassung = Object.entries(bericht)
|
||||
.filter(([, n]) => n > 0)
|
||||
.map(([blatt, n]) => `${n} ${blatt}`)
|
||||
.join(", ");
|
||||
await tx
|
||||
.insertInto("audit_log")
|
||||
.values({
|
||||
actor_user_id: akteur.userId,
|
||||
actor_name: akteur.name,
|
||||
action: "Import",
|
||||
target_label: "Massenimport",
|
||||
details: zusammenfassung || "nichts angelegt",
|
||||
})
|
||||
.execute();
|
||||
|
||||
return bericht;
|
||||
}
|
||||
Reference in New Issue
Block a user