
Ein System of Record (SoR) definiert pro Datendomäne eine führende Quelle und reduziert Abstimmungsaufwand. Eine sinnvolle Strukturierung für die Implementierung kann vier Bausteine umfassen: Datenspeicherung, Synchronisation, Validierung und Governance. Diese Anleitung zeigt, wie Sie ein SoR aufbauen und betreiben. system of record technische implementierung.
Kernaussagen
Eine sinnvolle Strukturierung für die Implementierung eines System of Record kann vier Bausteine umfassen: Datenspeicherung, Synchronisation, Validierung und Governance. Die maßgebliche Quelle für eine bestimmte Datendomäne – beispielsweise Belege, Zeiterfassungen oder Kontosalden – kann durch eine Datenbank mit Validierungsregeln, Synchronisationsmechanismen (APIs, Webhooks, Batch-Jobs) sowie Zugriffskontrolle und Governance-Prozesse realisiert werden.
Für die Implementierung wird empfohlen, pro Datendomäne einen klar benannten führenden Datensatz bzw. eine authoritative Quelle zu definieren und dies explizit zu dokumentieren – mehrere konkurrierende Quellen können zu Inkonsistenzen und Vertrauensverlust führen (IBM, 2024). Die Definition von Integrationsmustern ist ein Kernelement: vorgelagerte Systeme speisen das SoR, nachgelagerte Systeme konsumieren es primär – Klare Datenflüsse können die Nachvollziehbarkeit bei Audits unterstützen.
Was ist ein System of Record? Definition und Abgrenzung
Ein System of Record (SoR) ist das autorisierte, zentrale System, das die verbindliche Wahrheit für eine spezifische Datendomäne darstellt. Es ist die einzige Quelle, auf die sich Geschäftsentscheidungen stützen – nicht mehrere konkurrierende Systeme.
SoR vs. System of Engagement (SoE):
- SoR speichert und validiert Daten (z.B. Buchhaltungssystem für Belege, Clockify für Zeiterfassungen)
- SoE zeigt Daten an, ohne sie zu speichern (z.B. Dashboard, Reporting-Tool)
Praktische Beispiele:
- Buchhaltungssystem = SoR für Belege und Rechnungen
- Clockify = SoR für Zeiterfassungen und Projektzeiten
- Zentrales Data Warehouse = SoR für Reporting und Analysen
Technische Implementierung: Architektur-Grundlagen für System of Record
Die technische Implementierung eines System of Record unterscheidet sich grundlegend von der konzeptionellen Definition. Während die Definition festlegt, welches System die Quelle der Wahrheit ist, geht es bei der Implementierung um wie Sie diese Quelle technisch aufbauen, absichern und synchronisieren.
Vier zentrale Bereiche für ein zuverlässiges SoR:
| Baustein | Beschreibung | Beispiel |
|---|---|---|
| Datenspeicherung | Zentrale Datenbank, die alle Datendomänen speichert und Mandanten isoliert | PostgreSQL mit Row-Level Security, AWS RDS |
| Synchronisationsmechanismen | APIs, Webhooks und Batch-Jobs, die Daten zwischen Systemen austauschen | REST-API für Buchhaltungssystem-Integration, Webhook für Clockify-Updates |
| Validierungslogik | Regeln, die fehlerhafte oder inkonsistente Daten vor der Speicherung ausschließen | Datenbank-Constraints, Trigger, Application-Level-Validierung |
| Zugriffskontrolle & Governance | Authentifizierung, Autorisierung und Audit-Trails für alle Änderungen | Role-Based Access Control (RBAC), Audit-Logs, Compliance-Tracking |
Ohne diese Bereiche kann ein SoR nicht zuverlässig funktionieren. Typische Fehler: Datenbank ohne Backup-Strategie, Syncs ohne Fehlerbehandlung, Validierung nur auf Applikationsebene (nicht auf Datenbankebene), fehlende Audit-Trails. Die Folge: Datenverlust, Inkonsistenzen und keine Möglichkeit, Fehler nachzuvollziehen.
Praxisorientierte Checkliste: System of Record in 7 Schritten implementieren
Hinweis: Die folgende Checkliste stellt eine praxisbasierte Empfehlung dar, keinen Industriestandard.
Die technische Implementierung eines System of Record folgt einem strukturierten Prozess: Infrastruktur aufbauen, Datendomänen definieren, Integrationsflüsse konfigurieren und Governance etablieren. Diese Checkliste deckt alle sieben Schritte ab – von der Infrastruktur-Planung bis zum laufenden Monitoring.
1. Infrastruktur und Datenbankauswahl
Wählen Sie eine Datenbank, die Ihre Anforderungen erfüllt: Skalierbarkeit (Multi-Mandanten), Zuverlässigkeit (Backups, Failover), Performance (Abfragen auf Millionen von Datensätzen) und Sicherheit (Verschlüsselung, RLS). Typische Optionen: PostgreSQL (Open Source, robust), MySQL (weit verbreitet), Cloud-Datenbanken (AWS RDS, Google Cloud SQL, Azure SQL Database).
Checkliste:
- [ ] Datenbank-Engine ausgewählt (PostgreSQL, MySQL, Cloud-Lösung)
- [ ] Hochverfügbarkeit konfiguriert (Replikation, Failover-Mechanismen)
- [ ] Backup-Strategie definiert (täglich, mit Point-in-Time-Recovery)
- [ ] Verschlüsselung aktiviert (in Transit und at Rest)
- [ ] Disaster-Recovery-Plan dokumentiert
Ergebnis: Eine produktive Datenbankinstanz mit Backup- und Failover-Konfiguration.
2. Datenbankschema und Mandanten-Isolation
Entwerfen Sie ein Datenbankschema, das alle Datendomänen speichert und gleichzeitig Mandanten isoliert. Jede Tabelle sollte eine tenant_id-Spalte haben, um Daten pro Mandant zu partitionieren. Implementieren Sie Row-Level Security (RLS) auf Datenbankebene, damit Queries automatisch nach tenant_id gefiltert werden – das verhindert versehentliche Datenlecks.
Schema-Beispiel:
CREATE TABLE sor_belege (
id UUID PRIMARY KEY,
tenant_id UUID NOT NULL,
belegart VARCHAR(50) NOT NULL,
belegstatus VARCHAR(50) NOT NULL,
betrag DECIMAL(12, 2) NOT NULL,
erstellungsdatum TIMESTAMP NOT NULL,
letzte_aenderung TIMESTAMP NOT NULL,
CONSTRAINT fk_tenant FOREIGN KEY (tenant_id) REFERENCES tenants(id)
);
CREATE INDEX idx_tenant_belegart ON sor_belege(tenant_id, belegart);
-- Row-Level Security
ALTER TABLE sor_belege ENABLE ROW LEVEL SECURITY;
CREATE POLICY sor_belege_isolation ON sor_belege
USING (tenant_id = current_setting('app.current_tenant_id')::UUID);
Checkliste:
- [ ] Tabellen für alle Datendomänen erstellt (Belege, Zeiterfassungen, Abwesenheiten, Kontosalden)
- [ ]
tenant_idin jeder Tabelle vorhanden - [ ] Indizes auf
tenant_idund häufig abgefragte Spalten erstellt - [ ] Row-Level Security aktiviert und getestet
- [ ] Audit-Tabellen für alle Änderungen erstellt
Ergebnis: Ein Datenbankschema mit Mandanten-Isolation und Audit-Trail.
3. API-Schicht und Authentifizierung
Bauen Sie eine API-Schicht auf, die externe Systeme (ein anderer Anbieter, Clockify, Timebutler) mit dem SoR verbindet. Implementieren Sie OAuth 2.0 oder API-Keys für Authentifizierung. Jede API-Anfrage sollte den tenant_id enthalten und auf Autorisierung geprüft werden – ein Clockify-Webhook darf nur Zeiterfassungsdaten ändern, nicht Beleginformationen.
API-Endpoints (Beispiel):
POST /api/v1/sor/belege
- Erstellt einen neuen Beleg im SoR
- Authentifizierung: Bearer Token (OAuth 2.0)
- Autorisierung: Nur ein anderer Anbieter-Service darf schreiben
GET /api/v1/sor/belege?tenant_id=...&status=offen
- Ruft offene Belege ab
- Authentifizierung: API-Key oder OAuth
- Autorisierung: Nur der eigene Mandant sieht seine Daten
PATCH /api/v1/sor/belege/{id}
- Aktualisiert einen Beleg
- Authentifizierung: Bearer Token
- Autorisierung: Nur ein anderer Anbieter-Service darf schreiben
Checkliste:
- [ ] API-Framework ausgewählt (Express.js, FastAPI, Spring Boot)
- [ ] OAuth 2.0 oder API-Key-Authentifizierung implementiert
- [ ] Autorisierungsregeln pro Endpoint definiert
- [ ] Rate-Limiting konfiguriert (z. B. 1000 Requests pro Minute)
- [ ] API-Dokumentation (OpenAPI/Swagger) erstellt
Ergebnis: Eine produktive API mit Authentifizierung und Autorisierung.
4. Synchronisationsjobs und Fehlerbehandlung
Implementieren Sie Synchronisationsjobs, die Daten zwischen ein anderer Anbieter, Clockify, Timebutler und dem SoR austauschen. Nutzen Sie Webhooks für Echtzeit-Updates (wenn ein Beleg in ein anderer Anbieter erstellt wird, wird sofort eine Webhook-Nachricht an das SoR gesendet) und Batch-Jobs für regelmäßige Abgleiche (z. B. nächtlich um 2 Uhr).
Synchronisationsmuster:
ein anderer Anbieter → SoR (Echtzeit via Webhook)
- Beleg erstellt/aktualisiert in ein anderer Anbieter
- Webhook sendet Daten an SoR-API
- SoR speichert oder aktualisiert Beleg
- Bei Fehler: Retry mit exponentiellem Backoff
Clockify → SoR (Batch-Job, täglich um 3 Uhr)
- Job ruft alle Zeiterfassungen seit letztem Sync ab
- Vergleicht mit SoR-Daten
- Aktualisiert oder erstellt Einträge
- Bei Fehler: Benachrichtigung an Admin
Checkliste:
- [ ] Webhook-Endpoints für Echtzeit-Synchronisation konfiguriert
- [ ] Batch-Jobs für regelmäßige Abgleiche implementiert
- [ ] Retry-Logik mit exponentiellem Backoff eingebaut
- [ ] Fehlerbenachrichtigungen (E-Mail, Slack) konfiguriert
- [ ] Monitoring-Dashboard für Sync-Status erstellt
Ergebnis: Automatisierte Synchronisation mit Fehlerbehandlung.
5. Validierungsregeln und Datenqualität
Definieren Sie Validierungsregeln auf Datenbankebene (Constraints, Trigger) und Applikationsebene (API-Validierung). Beispiel: Ein Beleg darf nur die Status „offen", „bezahlt" oder „storniert" haben. Eine Zeiterfassung darf nicht in der Zukunft liegen. Ein Kontosaldo darf nicht negativ sein (außer bei Überziehungskrediten).
Validierungsbeispiel (PostgreSQL):
ALTER TABLE sor_belege
ADD CONSTRAINT chk_belegstatus
CHECK (belegstatus IN ('offen', 'bezahlt', 'storniert'));
ALTER TABLE sor_zeiterfassungen
ADD CONSTRAINT chk_datum_nicht_zukunft
CHECK (datum <= CURRENT_DATE);
Checkliste:
- [ ] Constraints auf Datenbankebene definiert
- [ ] Trigger für komplexe Validierungen erstellt
- [ ] API-Validierung implementiert (z. B. mit JSON Schema)
- [ ] Datenqualitäts-Reports erstellt (z. B. wöchentlich)
- [ ] Prozess für Bereinigung ungültiger Daten dokumentiert
Ergebnis: Validierte Daten mit hoher Qualität.
6. Governance und Change Management
Etablieren Sie Governance-Regeln: Wer darf welche Daten ändern? Wie werden Änderungen dokumentiert? Wer wird bei kritischen Änderungen benachrichtigt? Implementieren Sie Audit-Trails, die alle Änderungen protokollieren (wer, wann, was, warum).
Governance-Beispiel:
Regel 1: Nur ein anderer Anbieter-Service darf Belege erstellen/aktualisieren
Regel 2: Nur Clockify-Service darf Zeiterfassungen erstellen/aktualisieren
Regel 3: Nur Admin-Rolle darf Mandanten-Konfiguration ändern
Regel 4: Alle Änderungen werden in Audit-Tabelle protokolliert
Regel 5: Kritische Änderungen (z. B. Löschung von Belegen) erfordern Genehmigung
Checkliste:
- [ ] Governance-Regeln dokumentiert
- [ ] Audit-Trails für alle Änderungen implementiert
- [ ] Benachrichtigungen für kritische Änderungen konfiguriert
- [ ] Change-Management-Prozess definiert
- [ ] Regelmäßige Governance-Reviews geplant (z. B. quartalsweise)
Ergebnis: Nachvollziehbare Änderungen mit klaren Verantwortlichkeiten.
7. Monitoring und Alerting
Implementieren Sie Monitoring für alle kritischen Komponenten: Datenbank-Performance, API-Verfügbarkeit, Sync-Status, Datenqualität. Konfigurieren Sie Alerts, die Sie benachrichtigen, wenn etwas schiefgeht (z. B. Sync fehlgeschlagen, Datenbank-Latenz über 500ms, API-Fehlerrate über 1%).
Monitoring-Metriken:
- Datenbank-Latenz (p50, p95, p99)
- API-Verfügbarkeit (Uptime, Fehlerrate)
- Sync-Status (letzte erfolgreiche Synchronisation, Fehleranzahl)
- Datenqualität (Anzahl ungültiger Datensätze)
- Speicherplatz (Datenbank-Größe, Wachstumsrate)
Checkliste:
- [ ] Monitoring-Tool ausgewählt (Prometheus, Datadog, New Relic)
- [ ] Metriken für alle kritischen Komponenten konfiguriert
- [ ] Alerts für kritische Schwellwerte eingerichtet
- [ ] Dashboard für Echtzeit-Überwachung erstellt
- [ ] On-Call-Rotation für Incident-Response definiert
Ergebnis: Proaktives Monitoring mit schneller Fehlerreaktion.
---
Konkrete Implementierungsschritte für Multi-Mandanten-SaaS-Umgebungen
Für Multi-Mandanten-SaaS benötigen Sie drei Kernentscheidungen: Wählen Sie eine Isolationsstrategie (Shared Database mit Row-Level Security empfohlen), automatisieren Sie das Mandanten-Onboarding vollständig und implementieren Sie Partitionierung plus Rate-Limiting für Skalierung. Die folgenden Schritte zeigen die konkrete Umsetzung.
Schritt 1: Mandanten-Isolationsstrategie wählen
Es gibt drei gängige Strategien für Mandanten-Isolation:
1. Shared Database, Shared Schema: Alle Mandanten teilen sich dieselbe Datenbank und dasselbe Schema. Jede Tabelle hat eine tenant_id-Spalte. Vorteil: Einfach zu implementieren, kostengünstig. Nachteil: Risiko von Datenlecks, schwierig zu skalieren.
2. Shared Database, Separate Schemas: Alle Mandanten teilen sich dieselbe Datenbank, aber jeder Mandant hat sein eigenes Schema (z. B. tenant_123.belege, tenant_456.belege). Vorteil: Bessere Isolation, einfachere Backups pro Mandant. Nachteil: Komplexere Verwaltung, Schema-Migrationen müssen für alle Mandanten durchgeführt werden.
3. Separate Databases: Jeder Mandant hat seine eigene Datenbank. Vorteil: Maximale Isolation, einfache Skalierung pro Mandant. Nachteil: Hohe Kosten, komplexe Verwaltung.
Empfehlung: Shared Database, Shared Schema mit Row-Level Security (RLS). Diese Strategie bietet ein gutes Gleichgewicht zwischen Isolation, Kosten und Verwaltungsaufwand. RLS stellt sicher, dass Queries automatisch nach tenant_id gefiltert werden – selbst wenn ein Entwickler vergisst, die WHERE tenant_id =...-Klausel hinzuzufügen.
Schritt 2: Mandanten-Onboarding automatisieren
Wenn ein neuer Mandant hinzugefügt wird, müssen Sie:
1. Einen neuen Eintrag in der tenants-Tabelle erstellen
2. API-Keys oder OAuth-Credentials für den Mandanten generieren
3. Initiale Konfiguration (z. B. Ein anderer Anbieter-Instanz-URL, Clockify-Workspace-ID) speichern
4. Webhooks in ein anderer Anbieter/Clockify für den Mandanten registrieren
5. Initiale Synchronisation durchführen (alle bestehenden Belege/Zeiterfassungen importieren)
Automatisierungsskript (Pseudocode):
def onboard_tenant(tenant_name, ein anderer Anbieter_url, clockify_workspace_id):
# 1. Tenant erstellen
tenant_id = create_tenant(tenant_name)
# 2. API-Keys generieren
api_key = generate_api_key(tenant_id)
# 3. Konfiguration speichern
save_config(tenant_id, {
'ein anderer Anbieter_url': Ein anderer Anbieter_url,
'clockify_workspace_id': clockify_workspace_id
})
# 4. Webhooks registrieren
register_ein anderer Anbieter_webhook(tenant_id, ein anderer Anbieter_url)
register_clockify_webhook(tenant_id, clockify_workspace_id)
# 5. Initiale Synchronisation
sync_ein anderer Anbieter_belege(tenant_id)
sync_clockify_zeiterfassungen(tenant_id)
return tenant_id, api_key
Checkliste:
- [ ] Onboarding-Skript implementiert
- [ ] Automatische Webhook-Registrierung konfiguriert
- [ ] Initiale Synchronisation getestet
- [ ] Rollback-Prozess für fehlgeschlagenes Onboarding definiert
- [ ] Dokumentation für Mandanten-Onboarding erstellt
Schritt 3: Mandanten-spezifische Konfiguration verwalten
Jeder Mandant kann unterschiedliche Anforderungen haben: Mandant A nutzt ein anderer Anbieter und Clockify, Mandant B nutzt nur ein anderer Anbieter, Mandant C nutzt zusätzlich Timebutler. Speichern Sie diese Konfiguration in einer tenant_config-Tabelle:
CREATE TABLE tenant_config (
tenant_id UUID PRIMARY KEY,
ein anderer Anbieter_enabled BOOLEAN DEFAULT TRUE,
ein anderer Anbieter_url VARCHAR(255),
clockify_enabled BOOLEAN DEFAULT FALSE,
clockify_workspace_id VARCHAR(255),
timebutler_enabled BOOLEAN DEFAULT FALSE,
timebutler_api_key VARCHAR(255),
sync_interval_minutes INT DEFAULT 60,
CONSTRAINT fk_tenant FOREIGN KEY (tenant_id) REFERENCES tenants(id)
);
Checkliste:
- [ ]
tenant_config-Tabelle erstellt - [ ] Konfiguration über API abrufbar
- [ ] Änderungen an Konfiguration werden protokolliert
- [ ] Validierung für Konfigurationswerte implementiert
- [ ] UI für Mandanten-Konfiguration erstellt (optional)
Schritt 4: Skalierung und Performance-Optimierung
Wenn ein Mandant plötzlich 10x mehr Daten hat, darf das die Performance für andere Mandanten nicht beeinträchtigen. Implementieren Sie:
1. Partitionierung: Teilen Sie große Tabellen nach tenant_id auf (PostgreSQL unterstützt native Partitionierung). 2. Caching: Nutzen Sie Redis oder Memcached, um häufig abgefragte Daten zu cachen (z. B. Mandanten-Konfiguration). 3. Read Replicas: Nutzen Sie Read Replicas für Reporting-Queries, um die Last auf der primären Datenbank zu reduzieren. 4.
Rate-Limiting pro Mandant: Begrenzen Sie die Anzahl der API-Requests pro Mandant (z. B. 1000 Requests pro Minute).
Partitionierungsbeispiel (PostgreSQL):
CREATE TABLE sor_belege (
id UUID,
tenant_id UUID NOT NULL,
belegart VARCHAR(50) NOT NULL,
belegstatus VARCHAR(50) NOT NULL,
betrag DECIMAL(12, 2) NOT NULL,
erstellungsdatum TIMESTAMP NOT NULL,
letzte_aenderung TIMESTAMP NOT NULL,
PRIMARY KEY (tenant_id, id)
) PARTITION BY HASH (tenant_id);
CREATE TABLE sor_belege_p0 PARTITION OF sor_belege FOR VALUES WITH (MODULUS 4, REMAINDER 0); CREATE TABLE sor_belege_p1 PARTITION OF sor_belege FOR VALUES WITH (MODULUS 4, REMAINDER 1); CREATE TABLE sor_belege_p2 PARTITION OF sor_belege FOR VALUES WITH (MODULUS 4, REMAINDER 2); CREATE TABLE sor_belege_p3 PARTITION OF sor_belege FOR VALUES WITH (MODULUS 4,
REMAINDER 3); ```
**Checkliste:**
- [ ] Partitionierung für große Tabellen implementiert
- [ ] Caching-Layer konfiguriert
- [ ] Read Replicas eingerichtet
- [ ] Rate-Limiting pro Mandant aktiviert
- [ ] Performance-Tests mit realistischen Datenmengen durchgeführt
### Schritt 5: Mandanten-Offboarding und Datenlöschung
Wenn ein Mandant das System verlässt, müssen Sie alle seine Daten löschen (DSGVO-Compliance). Implementieren Sie einen Offboarding-Prozess:
1. Mandanten-Status auf „deaktiviert" setzen (sofortiger Zugriffsstopp)
2. Alle Webhooks deregistrieren
3. Alle Daten des Mandanten löschen (nach Ablauf der Aufbewahrungsfrist)
4. Audit-Log für Löschung erstellen
**Offboarding-Skript (Pseudocode):**
```python
def offboard_tenant(tenant_id):
## 1. Mandant deaktivieren
deactivate_tenant(tenant_id)
## 2. Webhooks deregistrieren
deregister_all_webhooks(tenant_id)
## 3. Daten löschen (nach Aufbewahrungsfrist)
schedule_data_deletion(tenant_id, days=30)
## 4. Audit-Log erstellen
log_audit_event(tenant_id, 'tenant_offboarded')
Checkliste:
- [ ] Offboarding-Skript implementiert
- [ ] Automatische Webhook-Deregistrierung konfiguriert
- [ ] Datenlöschung nach Aufbewahrungsfrist automatisiert
- [ ] Audit-Log für Offboarding erstellt
- [ ] DSGVO-Compliance überprüft
---
Datenfluss-Diagramme und Synchronisationslogik für operative Steuerung
Datenfluss-Diagramme visualisieren, wie Daten zwischen Systemen fließen. Sie zeigen, welches System das SoR ist, welche Systeme Daten einspeisen und welche Systeme Daten konsumieren. Ohne diese Visualisierung ist es schwierig, Datenkonflikte zu identifizieren oder Synchronisationsfehler zu debuggen.
Datenfluss-Diagramm: Ein anderer Anbieter als SoR für Belege
┌─────────────────┐
│ ein anderer Anbieter (SoR) │
│ für Belege │
└────────┬────────┘
│
│ Webhook (Echtzeit)
│ bei Beleg-Erstellung/-Änderung
▼
┌─────────────────┐
│ SoR-Datenbank │
│ (PostgreSQL) │
└────────┬────────┘
│
│ API-Abfrage
│ (GET /api/v1/sor/belege)
▼
┌─────────────────┐
│ Dashboard │
│ (System of │
│ Engagement) │
└─────────────────┘
Erklärung:
1. ein anderer Anbieter ist das SoR für Belege. Alle Belege werden in ein anderer Anbieter erstellt und verwaltet. 2. Wenn ein Beleg in ein anderer Anbieter erstellt oder geändert wird, sendet ein anderer Anbieter eine Webhook-Nachricht an die SoR-Datenbank. 3. Die SoR-Datenbank speichert den Beleg und validiert die Daten (z. B.
Belegstatus muss „offen", „bezahlt" oder „storniert" sein). 4. Das Dashboard (System of Engagement) ruft Belege über die API ab und zeigt sie an. Das Dashboard schreibt keine Daten zurück – es ist nur ein Konsument.
Datenfluss-Diagramm: Clockify als SoR für Zeiterfassungen
(System of │ │ Engagement) │ └─────────────────┘ ``` [Brutto Eingangsrechnung und SoR](https://centralcommand.de/blog/definition-brutto-eingangsrechnung-offener-posten-system-of-record)
**Erklärung:**
1. **Clockify** ist das SoR für Zeiterfassungen. Alle Zeiterfassungen werden in Clockify erstellt. 2. Ein **Batch-Job** läuft täglich um 3 Uhr und ruft alle Zeiterfassungen seit dem letzten Sync ab. 3. Die **SoR-Datenbank** speichert die Zeiterfassungen und validiert die Daten (z. B. Datum darf nicht in der Zukunft liegen). 4.
Das **Dashboard** ruft Zeiterfassungen über die API ab und zeigt sie an.
### Datenfluss-Diagramm: Bidirektionale Synchronisation (Timebutler ↔ SoR)
┌─────────────────┐
│ Timebutler │
│ (SoR für │
│ Abwesenheiten) │
└────────┬────────┘
│
│ Webhook (Echtzeit)
│ bei Abwesenheits-Erstellung/-Änderung
▼
┌─────────────────┐
│ SoR-Datenbank │
│ (PostgreSQL) │
└────────┬────────┘
│
│ Webhook (Echtzeit)
│ bei Abwesenheits-Genehmigung im Dashboard
▼
┌─────────────────┐
│ Timebutler │
│ (Status-Update) │
└─────────────────┘
**Erklärung:**
1. **Timebutler** ist das SoR für Abwesenheiten. Alle Abwesenheiten werden in Timebutler erstellt. 2. Wenn eine Abwesenheit in Timebutler erstellt wird, sendet Timebutler eine **Webhook-Nachricht** an die SoR-Datenbank. 3. Die **SoR-Datenbank** speichert die Abwesenheit. 4.
Wenn ein Manager im Dashboard eine Abwesenheit genehmigt, sendet die SoR-Datenbank eine **Webhook-Nachricht** zurück an Timebutler, um den Status zu aktualisieren.
**Achtung:** Bidirektionale Synchronisation ist komplex und fehleranfällig. Vermeiden Sie sie, wenn möglich. Wenn Sie sie benötigen, implementieren Sie Konfliktauflösungsregeln (z. B. „Timebutler-Status hat Vorrang bei Konflikten").
### Synchronisationslogik: Konfliktauflösung
Was passiert, wenn ein Beleg in ein anderer Anbieter und im SoR gleichzeitig geändert wird? Sie benötigen Konfliktauflösungsregeln:
**Regel 1: Last-Write-Wins (LWW)**
Der letzte Schreibvorgang gewinnt. Beispiel: Beleg wird in ein anderer Anbieter um 10:00 Uhr geändert, im SoR um 10:01 Uhr. Die Änderung um 10:01 Uhr überschreibt die Änderung um 10:00 Uhr.
**Regel 2: Source-of-Truth-Wins**
Das SoR hat immer Vorrang. Beispiel: Ein anderer Anbieter ist das SoR für Belege. Wenn ein Beleg in ein anderer Anbieter und im SoR gleichzeitig geändert wird, gewinnt die Änderung in ein anderer Anbieter.
**Regel 3: Manual Resolution**
Bei Konflikten wird ein Admin benachrichtigt, der manuell entscheidet, welche Änderung übernommen wird.
**Empfehlung:** Nutzen Sie **Source-of-Truth-Wins** für alle Datendomänen. Das ist die einfachste und zuverlässigste Regel.
### Synchronisationslogik: Fehlerbehandlung
Was passiert, wenn ein Webhook fehlschlägt? Implementieren Sie Retry-Logik mit exponentiellem Backoff:
```python
def send_webhook_with_retry(url, payload, max_retries=5):
for attempt in range(max_retries):
try:
response = requests.post(url, json=payload, timeout=10)
if response.status_code == 200:
return True
except requests.RequestException:
pass
# Exponentieller Backoff: 1s, 2s, 4s, 8s, 16s
time.sleep(2 ** attempt)
# Nach max_retries: Benachrichtigung an Admin
notify_admin(f"Webhook fehlgeschlagen nach {max_retries} Versuchen: {url}")
return False
Checkliste:
- [ ] Datenfluss-Diagramme für alle Datendomänen erstellt
- [ ] Konfliktauflösungsregeln dokumentiert
- [ ] Retry-Logik mit exponentiellem Backoff implementiert
- [ ] Fehlerbenachrichtigungen konfiguriert
- [ ] Monitoring für Synchronisationsfehler eingerichtet
---
Spezifische Checkliste für Zeiterfassung und Abwesenheitsmanagement-Integration
Die Integration von Zeiterfassungs- und Abwesenheitsmanagement-Systemen (Clockify, Timebutler) in ein SoR erfordert besondere Aufmerksamkeit: Zeiterfassungen müssen mit Projekten und Aufgaben verknüpft werden, Abwesenheiten müssen mit Urlaubsansprüchen abgeglichen werden, und Überstunden müssen korrekt berechnet werden. Diese Checkliste deckt alle kritischen Punkte ab.
1. Datenmodell für Zeiterfassungen
Definieren Sie ein Datenmodell, das alle relevanten Informationen speichert:
CONSTRAINT fk_tenant FOREIGN KEY (tenant_id) REFERENCES tenants(id), CONSTRAINT fk_mitarbeiter FOREIGN KEY (mitarbeiter_id) REFERENCES mitarbeiter(id), CONSTRAINT fk_projekt FOREIGN KEY (projekt_id) REFERENCES projekte(id), CONSTRAINT fk_aufgabe FOREIGN KEY (aufgabe_id) REFERENCES aufgaben(id), CONSTRAINT chk_datum_nicht_zukunft CHECK (datum <= CURRENT_DATE), CONSTRAINT chk_endzeit_nach_startzeit CHECK (endzeit > startzeit) );
CREATE INDEX idx_tenant_mitarbeiter_datum ON sor_zeiterfassungen(tenant_id, mitarbeiter_id, datum);
Checkliste:
- [ ] Tabelle
sor_zeiterfassungenerstellt - [ ] Fremdschlüssel zu
mitarbeiter,projekte,aufgabendefiniert - [ ] Constraints für Datenvalidierung hinzugefügt
- [ ] Indizes für häufige Abfragen erstellt
- [ ]
clockify_idfür Synchronisation mit Clockify gespeichert
2. Synchronisation mit Clockify
Implementieren Sie einen Batch-Job, der täglich alle Zeiterfassungen seit dem letzten Sync abruft:
## Projekt-ID aus Clockify-Project-ID mappen projekt_id = map_clockify_project_to_projekt(entry['projectId']) # Zeiterfassung im SoR speichern oder aktualisieren upsert_zeiterfassung( tenant_id=tenant_id, mitarbeiter_id=mitarbeiter_id, projekt_id=projekt_id, datum=entry['date'], startzeit=entry['start'], endzeit=entry['end'], dauer_minuten=entry['duration'], beschreibung=entry['description'], clockify_id=entry['id'] ) # Sync-Zeitpunkt aktualisieren update_last_sync_time(tenant_id, 'clockify', datetime.now) ```
**Checkliste:**
- [ ] Batch-Job für Clockify-Synchronisation implementiert
- [ ] Mapping zwischen Clockify-User-ID und Mitarbeiter-ID erstellt
- [ ] Mapping zwischen Clockify-Project-ID und Projekt-ID erstellt
- [ ] Upsert-Logik für Zeiterfassungen implementiert
- [ ] Fehlerbehandlung und Benachrichtigungen konfiguriert
### 3. Datenmodell für Abwesenheiten
Definieren Sie ein Datenmodell für Abwesenheiten:
```sql CREATE TABLE sor_abwesenheiten ( id UUID PRIMARY KEY, tenant_id UUID NOT NULL, mitarbeiter_id UUID NOT NULL, abwesenheitstyp VARCHAR(50) NOT NULL, -- 'urlaub', 'krankheit', 'sonderurlaub' startdatum DATE NOT NULL, enddatum DATE NOT NULL, anzahl_tage DECIMAL(4, 2) NOT NULL, status VARCHAR(50) NOT NULL, -- 'beantragt', 'genehmigt', 'abgelehnt' genehmiger_id UUID, genehmigungsdatum TIMESTAMP,
timebutler_id VARCHAR(255) UNIQUE, erstellungsdatum TIMESTAMP NOT NULL, letzte_aenderung TIMESTAMP NOT NULL, CONSTRAINT fk_tenant FOREIGN KEY (tenant_id) REFERENCES tenants(id), CONSTRAINT fk_mitarbeiter FOREIGN KEY (mitarbeiter_id) REFERENCES mitarbeiter(id), CONSTRAINT fk_genehmiger FOREIGN KEY (genehmiger_id) REFERENCES mitarbeiter(id), CONSTRAINT chk_enddatum_nach_startdatum CHECK (enddatum >= startdatum), CONSTRAINT chk_status CHECK (status IN ('beantragt', 'genehmigt', 'abgelehnt')) );
CREATE INDEX idx_tenant_mitarbeiter_datum ON sor_abwesenheiten(tenant_id, mitarbeiter_id, startdatum);
Checkliste:
- [ ] Tabelle
sor_abwesenheitenerstellt - [ ] Fremdschlüssel zu
mitarbeiterdefiniert - [ ] Constraints für Datenvalidierung hinzugefügt
- [ ] Indizes für häufige Abfragen erstellt
- [ ]
timebutler_idfür Synchronisation mit Timebutler gespeichert
4. Synchronisation mit Timebutler
Implementieren Sie Webhook-Endpoints für Echtzeit-Synchronisation:
Checkliste:
- [ ] Webhook-Endpoint für Timebutler-Abwesenheiten implementiert
- [ ] Webhook-Signatur-Verifizierung konfiguriert
- [ ] Mapping zwischen Timebutler-User-ID und Mitarbeiter-ID erstellt
- [ ] Upsert-Logik für Abwesenheiten implementiert
- [ ] Fehlerbehandlung und Benachrichtigungen konfiguriert
5. Überstundenberechnung
Implementieren Sie Logik zur Berechnung von Überstunden:
def calculate_overtime(tenant_id, mitarbeiter_id, monat):
## Sollarbeitszeit pro Monat abrufen (z. B.
160 Stunden) sollarbeitszeit = get_sollarbeitszeit(mitarbeiter_id, monat) # Tatsächliche Arbeitszeit aus Zeiterfassungen abrufen zeiterfassungen = get_zeiterfassungen(tenant_id, mitarbeiter_id, monat) istarbeitszeit = sum(z['dauer_minuten'] for z in zeiterfassungen) / 60 # Abwesenheiten abziehen (Urlaub, Krankheit) abwesenheiten = get_abwesenheiten(tenant_id, mitarbeiter_id, monat) abwesenheitstage = sum(a['anzahl_tage'] for a in abwesenheiten if a['status'] == 'genehmigt') abwesenheitsstunden = abwesenheitstage
* 8 # Annahme: 8 Stunden pro Tag # Überstunden berechnen ueberstunden = istarbeitszeit - (sollarbeitszeit - abwesenheitsstunden) return ueberstunden ```
**Checkliste:**
- [ ] Funktion zur Überstundenberechnung implementiert
- [ ] Sollarbeitszeit pro Mitarbeiter konfigurierbar
- [ ] Abwesenheiten werden korrekt abgezogen
- [ ] Überstunden-Report erstellt (z. B. monatlich)
- [ ] Benachrichtigungen bei hohen Überstunden konfiguriert
### 6. Urlaubsanspruch und Resturlaub
Implementieren Sie Logik zur Verwaltung von Urlaubsansprüchen:
```sql CREATE TABLE sor_urlaubsanspruch ( id UUID PRIMARY KEY, tenant_id UUID NOT NULL, mitarbeiter_id UUID NOT NULL, jahr INT NOT NULL, anspruch_tage DECIMAL(4, 2) NOT NULL, genommen_tage DECIMAL(4, 2) DEFAULT 0, resturlaub_tage DECIMAL(4, 2) GENERATED ALWAYS AS (anspruch_tage - genommen_tage) STORED, CONSTRAINT fk_tenant FOREIGN KEY (tenant_id) REFERENCES tenants(id), CONSTRAINT fk_mitarbeiter
FOREIGN KEY (mitarbeiter_id) REFERENCES mitarbeiter(id), UNIQUE (tenant_id, mitarbeiter_id, jahr) ); ```
**Checkliste:**
- [ ] Tabelle `sor_urlaubsanspruch` erstellt
- [ ] Automatische Berechnung von Resturlaub implementiert
- [ ] Urlaubsanspruch pro Mitarbeiter konfigurierbar
- [ ] Benachrichtigungen bei niedrigem Resturlaub konfiguriert
- [ ] Jahresabschluss-Prozess für Urlaubsübertrag dokumentiert
---
## Governance-Modelle für zentrale ein anderer Anbieter-Instanz-Verwaltung
Drei Governance-Modelle stehen zur Wahl: Zentralisierte Verwaltung (Admin-Team kontrolliert alles), delegierte Verwaltung (Mandanten-Admins haben Teilautonomie) oder Self-Service mit automatischen Compliance-Checks. Wählen Sie nach Mandantenzahl und Compliance-Anforderungen – Start-ups beginnen meist mit Modell 1, Enterprise-Umgebungen benötigen Modell 3.
### Governance-Modell 1: Zentralisierte Verwaltung (einfach)
In diesem Modell hat ein zentrales Admin-Team volle Kontrolle über alle ein anderer Anbieter-Instanzen. Alle Änderungen werden vom Admin-Team durchgeführt. Mandanten haben nur Lesezugriff auf ihre Daten.
**Rollen:**
- **Super-Admin**: Volle Kontrolle über alle Mandanten, kann Mandanten anlegen/löschen, Konfigurationen ändern
- **Mandanten-User**: Lesezugriff auf eigene Daten, kann keine Änderungen vornehmen
**Prozesse:**
- Mandanten-Onboarding: Super-Admin erstellt neuen Mandanten
- Konfigurationsänderungen: Super-Admin führt Änderungen durch
- Beleg-Löschung: Super-Admin löscht Belege nach Genehmigung
**Vorteile:**
- Einfach zu implementieren
- Hohe Kontrolle über alle Änderungen
- Geringes Risiko von Fehlkonfigurationen
**Nachteile:**
- Engpass beim Admin-Team
- Langsame Reaktionszeiten bei Änderungswünschen
- Geringe Autonomie für Mandanten
**Checkliste:**
- [ ] Rollen definiert und dokumentiert
- [ ] Zugriffskontrolle in API implementiert
- [ ] Prozesse für Onboarding/Offboarding dokumentiert
- [ ] Eskalationspfad für dringende Änderungen definiert
- [ ] Regelmäßige Governance-Reviews geplant
### Governance-Modell 2: Delegierte Verwaltung (mittel)
In diesem Modell haben Mandanten mehr Autonomie. Jeder Mandant hat einen Mandanten-Admin, der bestimmte Änderungen selbst durchführen kann (z. B. Konfiguration anpassen, Benutzer hinzufügen). Kritische Änderungen (z. B. Mandant löschen) erfordern weiterhin Genehmigung vom Super-Admin.
**Rollen:**
- **Super-Admin**: Volle Kontrolle über alle Mandanten, kann Mandanten anlegen/löschen
- **Mandanten-Admin**: Kann Konfiguration des eigenen Mandanten ändern, Benutzer hinzufügen/entfernen
- **Mandanten-User**: Lesezugriff auf eigene Daten, kann keine Änderungen vornehmen
**Prozesse:**
- Mandanten-Onboarding: Super-Admin erstellt neuen Mandanten, Mandanten-Admin übernimmt Konfiguration
- Konfigurationsänderungen: Mandanten-Admin führt Änderungen durch
- Beleg-Löschung: Mandanten-Admin kann Belege löschen (mit Audit-Log)
- Mandanten-Löschung: Erfordert Genehmigung vom Super-Admin
**Vorteile:**
- Höhere Autonomie für Mandanten
- Schnellere Reaktionszeiten bei Änderungswünschen
- Entlastung des Admin-Teams
**Nachteile:**
- Höheres Risiko von Fehlkonfigurationen
- Komplexere Zugriffskontrolle
- Mehr Schulungsaufwand für Mandanten-Admins
**Checkliste:**
- [ ] Rollen definiert und dokumentiert
- [ ] Zugriffskontrolle in API implementiert (RBAC)
- [ ] Genehmigungsprozess für kritische Änderungen definiert
- [ ] Schulungsmaterial für Mandanten-Admins erstellt
- [ ] Audit-Logs für alle Änderungen aktiviert
### Governance-Modell 3: Self-Service mit Compliance-Checks (komplex)
In diesem Modell haben Mandanten maximale Autonomie. Mandanten können selbst Konfigurationen ändern, Benutzer hinzufügen und sogar Mandanten anlegen (z. B. für Tochtergesellschaften). Alle Änderungen werden automatisch auf Compliance geprüft (z. B. DSGVO, Steuerrecht). Bei Verstößen werden Änderungen blockiert oder ein Admin wird benachrichtigt.
**Rollen:**
- **Super-Admin**: Volle Kontrolle über alle Mandanten, kann Compliance-Regeln definieren
- **Mandanten-Admin**: Kann Konfiguration des eigenen Mandanten ändern, Benutzer hinzufügen/entfernen, Sub-Mandanten anlegen
- **Mandanten-User**: Lesezugriff auf eigene Daten, kann keine Änderungen vornehmen
- **Compliance-Officer**: Kann Compliance-Regeln definieren, Audit-Logs einsehen
**Prozesse:**
- Mandanten-Onboarding: Mandanten-Admin kann selbst neue Sub-Mandanten anlegen
- Konfigurationsänderungen: Mandanten-Admin führt Änderungen durch, automatische Compliance-Checks
- Beleg-Löschung: Mandanten-Admin kann Belege löschen (mit Audit-Log und Compliance-Check)
- Compliance-Verstoß: Automatische Benachrichtigung an Compliance-Officer, Änderung wird blockiert
**Vorteile:**
- Maximale Autonomie für Mandanten
- Schnellste Reaktionszeiten
- Automatisierte Compliance-Checks
**Nachteile:**
- Höchste Komplexität
- Erfordert robuste Compliance-Engine
- Höherer Schulungsaufwand
**Checkliste:**
- [ ] Rollen definiert und dokumentiert
- [ ] Zugriffskontrolle in API implementiert (RBAC mit feingranularen Berechtigungen)
- [ ] Compliance-Engine implementiert (z. B. mit Regel-Engine)
- [ ] Automatische Benachrichtigungen bei Compliance-Verstößen konfiguriert
- [ ] Schulungsmaterial für Mandanten-Admins und Compliance-Officers erstellt
### Compliance-Regeln für ein anderer Anbieter-Integration
Definieren Sie Compliance-Regeln, die automatisch geprüft werden:
**Regel 1: DSGVO – Datenlöschung nach Aufbewahrungsfrist**
Belege müssen nach Ablauf der gesetzlichen Aufbewahrungsfrist (10 Jahre) gelöscht werden.
```python
def check_dsgvo_compliance(tenant_id):
belege = get_belege(tenant_id)
for beleg in belege:
if beleg['erstellungsdatum'] < datetime.now - timedelta(days=3650):
## Beleg ist älter als 10 Jahre
notify_compliance_officer(f"Beleg {beleg['id']} muss gelöscht werden")
Regel 2: Steuerrecht – Belege dürfen nicht geändert werden
Belege dürfen nach Erstellung nicht mehr geändert werden (außer Stornierung).
def check_steuerrecht_compliance(beleg_id, changes):
beleg = get_beleg(beleg_id)
if beleg['status'] == 'bezahlt' and 'betrag' in changes:
## Beleg ist bezahlt und Betrag soll geändert werden
raise ComplianceError("Bezahlte Belege dürfen nicht geändert werden")
Regel 3: Interne Richtlinie – Belege über 10.000 EUR erfordern Genehmigung
Belege über 10.000 EUR müssen von einem Manager genehmigt werden.
def check_internal_policy_compliance(beleg):
if beleg['betrag'] > 10000 and beleg['status']!= 'genehmigt':
notify_manager(f"Beleg {beleg['id']} erfordert Genehmigung")
Checkliste:
- [ ] Compliance-Regeln definiert und dokumentiert
- [ ] Automatische Compliance-Checks implementiert
- [ ] Benachrichtigungen bei Compliance-Verstößen konfiguriert
- [ ] Regelmäßige Compliance-Audits geplant
- [ ] Schulungsmaterial für Compliance-Regeln erstellt
---
Best Practices für System of Record Implementierung
Die folgenden Best Practices basieren auf Erfahrungen aus realen Implementierungen und helfen Ihnen, häufige Fehler zu vermeiden.
1. Starten Sie mit einem Minimum Viable Product (MVP)
Implementieren Sie nicht alle Datendomänen gleichzeitig. Starten Sie mit einer kritischen Datendomäne (z. B. Belege) und erweitern Sie schrittweise. Das reduziert Komplexität und ermöglicht schnelles Lernen.
MVP-Umfang:
- Eine Datendomäne (z. B. Belege)
- Ein SoR (z. B. Ein anderer Anbieter)
- Ein Konsument (z. B. Dashboard)
- Echtzeit-Synchronisation via Webhook
- Basis-Validierung und Fehlerbehandlung
2. Dokumentieren Sie alles
Dokumentation ist kritisch für langfristigen Erfolg. Dokumentieren Sie:
- Datenfluss-Diagramme
- API-Spezifikationen
- Synchronisationsregeln
- Governance-Prozesse
- Fehlerbehandlung und Eskalationspfade
Nutzen Sie Tools wie Confluence, Notion oder GitHub Wiki für zentrale Dokumentation.
3. Implementieren Sie Monitoring von Anfang an
Monitoring ist nicht optional. Implementieren Sie Monitoring für:
- Datenbank-Performance
- API-Verfügbarkeit
- Sync-Status
- Datenqualität
- Fehlerrate
Nutzen Sie Tools wie Prometheus, Datadog oder New Relic.
4. Testen Sie mit realistischen Datenmengen
Testen Sie nicht nur mit 100 Datensätzen. Testen Sie mit realistischen Datenmengen (z. B. 1 Million Belege, 10.000 Zeiterfassungen pro Tag). Das deckt Performance-Probleme frühzeitig auf.
5. Planen Sie für Skalierung
Planen Sie von Anfang an für Skalierung:
- Nutzen Sie Partitionierung für große Tabellen
- Nutzen Sie Caching für häufig abgefragte Daten
- Nutzen Sie Read Replicas für Reporting-Queries
- Implementieren Sie Rate-Limiting pro Mandant
6. Etablieren Sie klare Governance-Regeln
Definieren Sie klare Governance-Regeln:
- Wer darf welche Daten ändern?
- Wie werden Änderungen genehmigt?
- Wie werden Compliance-Verstöße behandelt?
Dokumentieren Sie diese Regeln und schulen Sie Ihr Team.
7. Automatisieren Sie alles
Automatisieren Sie so viel wie möglich:
- Mandanten-Onboarding
- Synchronisation
- Backups
- Compliance-Checks
- Fehlerbenachrichtigungen
Manuelle Prozesse sind fehleranfällig und nicht skalierbar.
---
Datenbank-Design und Mandanten-Isolation: Multi-Instanzen-Verwaltung
Das Datenbank-Design ist das Fundament Ihres SoR. Ein schlechtes Design führt zu Performance-Problemen, Datenlecks und hohen Wartungskosten. Dieser Abschnitt beschreibt Best Practices für Datenbank-Design und Mandanten-Isolation.
Mandanten-Isolation: Shared Database, Shared Schema mit RLS
Die empfohlene Strategie für Multi-Mandanten-SaaS ist Shared Database, Shared Schema mit Row-Level Security (RLS). Alle Mandanten teilen sich dieselbe Datenbank und dasselbe Schema, aber jede Zeile hat eine tenant_id-Spalte. RLS stellt sicher, dass Queries automatisch nach tenant_id gefiltert werden.
Vorteile:
- Kostengünstig (eine Datenbank für alle Mandanten)
- Einfach zu skalieren (Partitionierung nach
tenant_id) - Automatische Isolation durch RLS (kein Risiko von Datenlecks)
Nachteile:
- Komplexere Backup-Strategie (alle Mandanten in einem Backup)
- Schwieriger, einzelne Mandanten zu migrieren
Implementierung (PostgreSQL):
-- Tabelle mit tenant_id
CREATE TABLE sor_belege (
id UUID PRIMARY KEY,
tenant_id UUID NOT NULL,
belegart VARCHAR(50) NOT NULL,
belegstatus VARCHAR(50) NOT NULL,
betrag DECIMAL(12, 2) NOT NULL,
erstellungsdatum TIMESTAMP NOT NULL,
letzte_aenderung TIMESTAMP NOT NULL,
CONSTRAINT fk_tenant FOREIGN KEY (tenant_id) REFERENCES tenants(id)
);
-- Row-Level Security aktivieren
ALTER TABLE sor_belege ENABLE ROW LEVEL SECURITY;
-- Policy erstellen: Nur Zeilen mit eigenem tenant_id sichtbar
CREATE POLICY sor_belege_isolation ON sor_belege
USING (tenant_id = current_setting('app.current_tenant_id')::UUID);
-- In Applikation: tenant_id setzen vor jedem Query
SET app.current_tenant_id = 'tenant-123-uuid';
SELECT * FROM sor_belege; -- Zeigt nur Belege von tenant-123-uuid
Partitionierung für Skalierung
Wenn Ihre Tabellen sehr groß werden (z. B. 10 Millionen Belege), nutzen Sie Partitionierung. PostgreSQL unterstützt native Partitionierung nach tenant_id:
CREATE TABLE sor_belege (
id UUID,
tenant_id UUID NOT NULL,
belegart VARCHAR(50) NOT NULL,
belegstatus VARCHAR(50) NOT NULL,
betrag DECIMAL(12, 2) NOT NULL,
erstellungsdatum TIMESTAMP NOT NULL,
letzte_aenderung TIMESTAMP NOT NULL,
PRIMARY KEY (tenant_id, id)
) PARTITION BY HASH (tenant_id);
-- 4 Partitionen erstellen
CREATE TABLE sor_belege_p0 PARTITION OF sor_belege FOR VALUES WITH (MODULUS 4, REMAINDER 0);
CREATE TABLE sor_belege_p1 PARTITION OF sor_belege FOR VALUES WITH (MODULUS 4, REMAINDER 1);
CREATE TABLE sor_belege_p2 PARTITION OF sor_belege FOR VALUES WITH (MODULUS 4, REMAINDER 2);
CREATE TABLE sor_belege_p3 PARTITION OF sor_belege FOR VALUES WITH (MODULUS 4, REMAINDER 3);
Vorteile:
- Queries werden schneller (nur relevante Partition wird gescannt)
- Einfacher, alte Daten zu archivieren (Partition droppen)
- Bessere Parallelisierung bei großen Queries
Audit-Trails für Nachvollziehbarkeit
Implementieren Sie Audit-Trails, die alle Änderungen protokollieren:
CREATE TABLE sor_audit_log (
id UUID PRIMARY KEY,
tenant_id UUID NOT NULL,
tabelle VARCHAR(100) NOT NULL,
datensatz_id UUID NOT NULL,
aktion VARCHAR(50) NOT NULL, -- 'INSERT', 'UPDATE', 'DELETE'
benutzer_id UUID NOT NULL,
aenderungsdatum TIMESTAMP NOT NULL,
alte_werte JSONB,
neue_werte JSONB,
CONSTRAINT fk_tenant FOREIGN KEY (tenant_id) REFERENCES tenants(id)
);
-- Trigger für automatisches Audit-Logging
CREATE OR REPLACE FUNCTION audit_trigger_func
RETURNS TRIGGER AS $
BEGIN
INSERT INTO sor_audit_log (id, tenant_id, tabelle, datensatz_id, aktion, benutzer_id, aenderungsdatum, alte_werte, neue_werte)
VALUES (
gen_random_uuid,
NEW.tenant_id,
TG_TABLE_NAME,
NEW.id,
TG_OP,
current_setting('app.current_user_id')::UUID,
NOW,
CASE WHEN TG_OP = 'UPDATE' THEN row_to_json(OLD) ELSE NULL END,
row_to_json(NEW)
);
RETURN NEW;
END;
$ LANGUAGE plpgsql;
-- Trigger auf Tabelle anwenden
CREATE TRIGGER sor_belege_audit_trigger
AFTER INSERT OR UPDATE OR DELETE ON sor_belege
FOR EACH ROW EXECUTE FUNCTION audit_trigger_func;
Checkliste:
- [ ] RLS für alle Tabellen aktiviert
- [ ] Partitionierung für große Tabellen implementiert
- [ ] Audit-Trails für alle Änderungen aktiviert
- [ ] Indizes für häufige Abfragen erstellt
- [ ] Backup-Strategie definiert und getestet
---
Synchronisationsmechanismen: API-Integration, Webhooks und Fehlerbehandlung
Synchronisationsmechanismen sind das Herzstück Ihres SoR. Sie stellen sicher, dass Daten zwischen Systemen konsistent bleiben. Dieser Abschnitt beschreibt drei Synchronisationsmuster: Webhooks (Echtzeit), Batch-Jobs (regelmäßig) und Polling (fallback).
Synchronisationsmuster 1: Webhooks (Echtzeit)
Webhooks sind die bevorzugte Methode für Echtzeit-Synchronisation. Wenn ein Beleg in ein anderer Anbieter erstellt wird, sendet ein anderer Anbieter sofort eine Webhook-Nachricht an Ihr SoR.
Implementierung (ein anderer Anbieter → SoR):
@app.post("/webhooks/ein anderer Anbieter/beleg")
def ein anderer Anbieter_beleg_webhook(request: Request):
## 1. Webhook-Signatur verifizieren
signature = request.headers.get('X-ein anderer Anbieter-Signature')
if not verify_webhook_signature(request.body, signature):
return {"error": "Invalid signature"}, 401
## 2. Payload parsen
payload = request.json
tenant_id = payload['tenant_id']
beleg_data = payload['beleg']
## 3.
Beleg im SoR speichern oder aktualisieren
try:
upsert_beleg(
tenant_id=tenant_id,
belegart=beleg_data['art'],
belegstatus=beleg_data['status'],
betrag=beleg_data['betrag'],
erstellungsdatum=beleg_data['datum'],
ein anderer Anbieter_id=beleg_data['id']
)
return {"status": "success"}
except Exception as e:
## 4. Fehlerbehandlung
log_error(f"Webhook fehlgeschlagen: {e}")
notify_admin(f"Webhook fehlgeschlagen für Tenant {tenant_id}")
return {"error": str(e)}, 500
Webhook-Secret (geteilt zwischen ein anderer Anbieter und SoR)
import hmac
import hashlib
def verify_webhook_signature(payload, signature):
secret = get_webhook_secret # Aus Umgebungsvariable oder Secrets Manager
expected_signature = hmac.new(
secret.encode,
payload,
hashlib.sha256
).hexdigest
return hmac.compare_digest(expected_signature, signature)
Checkliste:
- [ ] Webhook-Endpoint implementiert
- [ ] Webhook-Signatur-Verifizierung konfiguriert
- [ ] Fehlerbehandlung und Logging implementiert
- [ ] Retry-Logik für fehlgeschlagene Webhooks konfiguriert
- [ ] Monitoring für Webhook-Fehlerrate eingerichtet
Synchronisationsmuster 2: Batch-Jobs (regelmäßig)
Batch-Jobs sind nützlich für Systeme, die keine Webhooks unterstützen (z. B. Clockify). Ein Batch-Job läuft regelmäßig (z. B. täglich um 3 Uhr) und ruft alle Änderungen seit dem letzten Sync ab.
Implementierung (Clockify → SoR):
def sync_clockify_batch_job:
tenants = get_all_tenants
for tenant in tenants:
try:
sync_clockify_zeiterfassungen(tenant['id'])
except Exception as e:
log_error(f"Batch-Job fehlgeschlagen für Tenant {tenant['id']}: {e}")
notify_admin(f"Batch-Job fehlgeschlagen für Tenant {tenant['id']}")
def sync_clockify_zeiterfassungen(tenant_id):
config = get_tenant_config(tenant_id)
if not config['clockify_enabled']:
return
## Letzter Sync-Zeitpunkt abrufen
last_sync = get_last_sync_time(tenant_id, 'clockify')
## Alle Zeiterfassungen seit letztem Sync abrufen
time_entries = clockify_api.get_time_entries(
workspace_id=config['clockify_workspace_id'],
api_key=config['clockify_api_key'],
start_date=last_sync
)
## Zeiterfassungen im SoR speichern
for entry in time_entries:
upsert_zeiterfassung(tenant_id, entry)
## Sync-Zeitpunkt aktualisieren
update_last_sync_time(tenant_id, 'clockify', datetime.now)
Checkliste:
- [ ] Batch-Job implementiert (z. B. mit Celery, Airflow)
- [ ] Cron-Schedule konfiguriert (z. B. täglich um 3 Uhr)
- [ ] Fehlerbehandlung und Logging implementiert
- [ ] Benachrichtigungen bei fehlgeschlagenen Jobs konfiguriert
- [ ] Monitoring für Job-Laufzeit und Fehlerrate eingerichtet
Synchronisationsmuster 3: Polling (Fallback)
Polling ist die einfachste, aber ineffizienteste Methode. Das SoR fragt regelmäßig (z. B. alle 5 Minuten) bei ein anderer Anbieter nach, ob es Änderungen gibt.
Implementierung (SoR → ein anderer Anbieter):
def poll_ein anderer Anbieter_for_changes:
tenants = get_all_tenants
for tenant in tenants:
try:
poll_ein anderer Anbieter_belege(tenant['id'])
except Exception as e:
log_error(f"Polling fehlgeschlagen für Tenant {tenant['id']}: {e}")
def poll_ein anderer Anbieter_belege(tenant_id):
config = get_tenant_config(tenant_id)
if not config['ein anderer Anbieter_enabled']:
return
last_sync = get_last_sync_time(tenant_id, 'ein anderer Anbieter')
## Alle Belege seit letztem Sync abrufen
belege = ein anderer Anbieter_api.get_belege(
url=config['ein anderer Anbieter_url'],
api_key=config['ein anderer Anbieter_api_key'],
since=last_sync
)
## Belege im SoR speichern
for beleg in belege:
upsert_beleg(tenant_id, beleg)
update_last_sync_time(tenant_id, 'ein anderer Anbieter', datetime.now)
Checkliste:
- [ ] Polling-Job implementiert
- [ ] Polling-Intervall konfiguriert (z. B. alle 5 Minuten)
- [ ] Fehlerbehandlung und Logging implementiert
- [ ] Rate-Limiting beachtet (ein anderer Anbieter-API hat Limits)
- [ ] Monitoring für Polling-Fehlerrate eingerichtet
Fehlerbehandlung: Retry mit exponentiellem Backoff
Wenn ein Webhook oder Batch-Job fehlschlägt, implementieren Sie Retry-Logik mit exponentiellem Backoff:
nach {max_retries} Versuchen: {url}") return False ```
**Checkliste:**
- [ ] Retry-Logik mit exponentiellem Backoff implementiert
- [ ] Maximale Anzahl von Retries konfiguriert
- [ ] Benachrichtigungen bei dauerhaften Fehlern konfiguriert
- [ ] Dead-Letter-Queue für fehlgeschlagene Nachrichten implementiert
- [ ] Monitoring für Retry-Rate eingerichtet
---
## Fazit und nächste Schritte
Als nächsten Schritt prüfen Sie, welche der oben genannten Punkte in Ihrem Setup schon greifen, und definieren Sie pro offenem Thema eine messbare Maßnahme.
## FAQ: Häufig gestellte Fragen zu System of Record Implementierung
### Was ist der Unterschied zwischen System of Record und Data Warehouse?
Ein System of Record (SoR) ist die maßgebliche Quelle für operative Daten – beispielsweise Belege in ein anderer Anbieter oder Zeiterfassungen in Clockify. Ein Data Warehouse ist eine zentrale Datenbank für analytische Daten, die aus mehreren SoRs aggregiert werden.
Das SoR ist für den täglichen Betrieb zuständig, das Data Warehouse für Reporting und Business Intelligence.
### Kann ich mehrere Systeme als SoR für dieselbe Datendomäne haben?
Nein. Pro Datendomäne sollte es genau ein SoR geben. Mehrere konkurrierende Quellen führen zu Inkonsistenzen und Vertrauensverlust (IBM, 2024). Wenn Sie mehrere Systeme haben, die dieselben Daten speichern, definieren Sie eines als SoR und synchronisieren Sie die anderen Systeme mit diesem SoR.
### Wie oft sollte ich Daten zwischen Systemen synchronisieren?
Das hängt von Ihren Anforderungen ab. Für kritische Daten (z. B. Belege) empfehlen wir Echtzeit-Synchronisation via Webhooks. Für weniger kritische Daten (z. B. Zeiterfassungen) reicht ein täglicher Batch-Job. Vermeiden Sie zu häufige Synchronisationen (z. B. alle 5 Minuten), da dies die API-Limits überschreiten und Kosten erhöhen kann.
### Was passiert, wenn ein Webhook fehlschlägt?
Implementieren Sie Retry-Logik mit exponentiellem Backoff. Wenn der Webhook nach mehreren Versuchen immer noch fehlschlägt, benachrichtigen Sie einen Admin. Speichern Sie fehlgeschlagene Webhooks in einer Dead-Letter-Queue, damit Sie sie später manuell verarbeiten können.
### Wie stelle ich sicher, dass meine Daten DSGVO-konform sind?
Implementieren Sie automatische Compliance-Checks: Belege müssen nach Ablauf der gesetzlichen Aufbewahrungsfrist (10 Jahre) gelöscht werden. Personenbezogene Daten müssen auf Anfrage gelöscht werden können. Nutzen Sie Audit-Trails, um alle Änderungen nachzuvollziehen. Schulen Sie Ihr Team regelmäßig zu DSGVO-Anforderungen.
### Wie skaliere ich mein SoR, wenn die Datenmengen wachsen?
Nutzen Sie Partitionierung für große Tabellen, Caching für häufig abgefragte Daten und Read Replicas für Reporting-Queries. Implementieren Sie Rate-Limiting pro Mandant, um zu verhindern, dass ein Mandant die Performance für alle anderen beeinträchtigt. Überwachen Sie die Datenbank-Performance regelmäßig und skalieren Sie proaktiv.
### Kann ich ein SoR auch für unstrukturierte Daten (z. B. Dokumente) verwenden?
Ja, aber mit Einschränkungen. Für unstrukturierte Daten (z. B. PDF-Rechnungen) empfehlen wir einen Object Storage (z. B. AWS S3, Google Cloud Storage) als SoR. Speichern Sie Metadaten (z. B. Dateiname, Upload-Datum, Mandant-ID) in Ihrer relationalen Datenbank und verlinken Sie auf die Dateien im Object Storage.
### Wie teste ich mein SoR vor dem Go-Live?
Testen Sie mit realistischen Datenmengen (z. B. 1 Million Belege, 10.000 Zeiterfassungen pro Tag). Führen Sie Last-Tests durch, um Performance-Probleme frühzeitig zu identifizieren. Testen Sie Fehlerszenarien (z. B. Webhook fehlgeschlagen, Datenbank nicht erreichbar). Führen Sie ein Disaster-Recovery-Szenario durch, um sicherzustellen, dass Sie Daten wiederherstellen können.
### Wie dokumentiere ich mein SoR für mein Team?
Erstellen Sie ein zentrales Dokumentations-Repository (z. B. Confluence, Notion, GitHub Wiki). Dokumentieren Sie Datenfluss-Diagramme, API-Spezifikationen, Synchronisationsregeln, Governance-Prozesse und Fehlerbehandlung. Halten Sie die Dokumentation aktuell und schulen Sie neue Teammitglieder regelmäßig.
## Quellen
- https://www.ivanti.com/de/glossary/system-of-record
- https://www.ibm.com/think/topics/system-of-record
---
*Haftungsausschluss / Disclaimer – Keine Rechtsberatung*
Die auf dieser Website / in diesem Dokument bereitgestellten Informationen dienen ausschließlich allgemeinen Informationszwecken. Sie stellen keine Rechtsberatung dar und können eine individuelle rechtliche Beratung durch einen qualifizierten Rechtsanwalt nicht ersetzen.
Obwohl die Inhalte mit größtmöglicher Sorgfalt erstellt wurden, wird keine Gewähr für die Richtigkeit, Vollständigkeit und Aktualität der bereitgestellten Informationen übernommen. Die Nutzung der Inhalte erfolgt auf eigene Gefahr des Nutzers.
Zwischen dem Anbieter dieser Informationen und dem Nutzer entsteht durch die Nutzung dieser Inhalte kein Mandatsverhältnis und keine anwaltliche Beratungsbeziehung.
Für die Klärung individueller Rechtsfragen wenden Sie sich bitte an einen zugelassenen Rechtsanwalt Ihres Vertrauens. Eine Haftung für Schäden, die durch die Nutzung oder Nichtnutzung der dargebotenen Informationen entstehen, ist – soweit gesetzlich zulässig – ausgeschlossen.