256 lines
13 KiB
Go
256 lines
13 KiB
Go
package audit
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
type PostgresRepository struct{ pool *pgxpool.Pool }
|
|
|
|
func NewPostgresRepository(pool *pgxpool.Pool) *PostgresRepository {
|
|
return &PostgresRepository{pool: pool}
|
|
}
|
|
|
|
func (r *PostgresRepository) Health(ctx context.Context) error {
|
|
var ok bool
|
|
return r.pool.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name='ai_calls')`).Scan(&ok)
|
|
}
|
|
|
|
func (r *PostgresRepository) UpsertCall(ctx context.Context, c CallRecord) error {
|
|
if c.StartedAt.IsZero() {
|
|
c.StartedAt = time.Now().UTC()
|
|
}
|
|
meta := jsonb(c.Metadata)
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO ai_calls(call_id,asterisk_channel_id,route,caller_number_masked,language,region_code,state,started_at,metadata)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)
|
|
ON CONFLICT(call_id) DO UPDATE SET asterisk_channel_id=EXCLUDED.asterisk_channel_id, route=EXCLUDED.route, caller_number_masked=EXCLUDED.caller_number_masked, language=EXCLUDED.language, region_code=EXCLUDED.region_code, state=EXCLUDED.state, updated_at=now(), metadata=EXCLUDED.metadata`, c.CallID, c.AsteriskChannelID, nonEmpty(c.Route, "unknown"), c.CallerNumberMasked, c.Language, c.RegionCode, c.State, c.StartedAt, meta)
|
|
return err
|
|
}
|
|
func (r *PostgresRepository) EndCall(ctx context.Context, callID string, reason string) error {
|
|
_, err := r.pool.Exec(ctx, `UPDATE ai_calls SET ended_at=now(), duration_ms=GREATEST(0, EXTRACT(EPOCH FROM (now()-started_at))*1000)::bigint, end_reason=$2, state='ENDED', updated_at=now() WHERE call_id=$1`, callID, reason)
|
|
return err
|
|
}
|
|
func (r *PostgresRepository) AddEvent(ctx context.Context, e EventRecord) error {
|
|
if e.CreatedAt.IsZero() {
|
|
e.CreatedAt = time.Now().UTC()
|
|
}
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO ai_call_events(call_id,event_type,event_source,state_before,state_after,severity,message,reason_code,created_at,metadata) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)`, e.CallID, e.EventType, e.EventSource, e.StateBefore, e.StateAfter, nonEmpty(e.Severity, "info"), e.Message, e.ReasonCode, e.CreatedAt, jsonb(e.Metadata))
|
|
return err
|
|
}
|
|
func (r *PostgresRepository) AddTranscript(ctx context.Context, t TranscriptRecord) error {
|
|
if t.CreatedAt.IsZero() {
|
|
t.CreatedAt = time.Now().UTC()
|
|
}
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO ai_transcript_events(call_id,speaker,event_type,language,text_redacted,text_hash,char_count,redaction_applied,created_at,metadata) VALUES($1,$2,$3,$4,$5,encode(digest($5,'sha256'),'hex'),$6,true,$7,$8)`, t.CallID, t.Speaker, t.EventType, t.Language, t.Text, len([]rune(t.Text)), t.CreatedAt, jsonb(t.Metadata))
|
|
return err
|
|
}
|
|
func (r *PostgresRepository) AddToolAudit(ctx context.Context, t ToolAuditRecord) error {
|
|
if t.CreatedAt.IsZero() {
|
|
t.CreatedAt = time.Now().UTC()
|
|
}
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO ai_tool_audit(call_id,tool_call_id,tool_name,state,language,region_code,allowed,denied,reason_code,args_redacted,result_redacted,duration_ms,created_at,metadata) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)`, t.CallID, t.ToolCallID, t.ToolName, t.State, t.Language, t.RegionCode, t.Allowed, t.Denied, t.ReasonCode, jsonb(t.Args), jsonb(t.Result), t.DurationMS, t.CreatedAt, jsonb(t.Metadata))
|
|
return err
|
|
}
|
|
func (r *PostgresRepository) AddKBAudit(ctx context.Context, k KBAuditRecord) error {
|
|
if k.CreatedAt.IsZero() {
|
|
k.CreatedAt = time.Now().UTC()
|
|
}
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO ai_kb_audit(call_id,query_redacted,language,region_code,result_count,top_score,cross_language_fallback_used,citations_count,no_answer,duration_ms,created_at,metadata) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)`, k.CallID, k.Query, k.Language, k.RegionCode, k.ResultCount, k.TopScore, k.CrossLanguageFallbackUsed, k.CitationsCount, k.NoAnswer, k.DurationMS, k.CreatedAt, jsonb(k.Metadata))
|
|
return err
|
|
}
|
|
func (r *PostgresRepository) AddHandoffAudit(ctx context.Context, h HandoffAuditRecord) error {
|
|
if h.CreatedAt.IsZero() {
|
|
h.CreatedAt = time.Now().UTC()
|
|
}
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO ai_handoff_audit(call_id,handoff_id,mode,status,reason_code,transfer_attempted,transfer_succeeded,target_redacted,summary_redacted,created_at,metadata) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)`, h.CallID, h.HandoffID, h.Mode, h.Status, h.ReasonCode, h.TransferAttempted, h.TransferSucceeded, h.Target, h.Summary, h.CreatedAt, jsonb(h.Metadata))
|
|
return err
|
|
}
|
|
func (r *PostgresRepository) AddProviderAudit(ctx context.Context, p ProviderAuditRecord) error {
|
|
if p.CreatedAt.IsZero() {
|
|
p.CreatedAt = time.Now().UTC()
|
|
}
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO ai_provider_audit(call_id,provider,event_type,severity,error_redacted,input_audio_bytes,output_audio_bytes,events_received,events_sent,created_at,metadata) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)`, p.CallID, p.Provider, p.EventType, nonEmpty(p.Severity, "info"), p.Error, p.InputAudioBytes, p.OutputAudioBytes, p.EventsReceived, p.EventsSent, p.CreatedAt, jsonb(p.Metadata))
|
|
return err
|
|
}
|
|
func (r *PostgresRepository) AddMediaAudit(ctx context.Context, m MediaAuditRecord) error {
|
|
if m.CreatedAt.IsZero() {
|
|
m.CreatedAt = time.Now().UTC()
|
|
}
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO ai_media_audit(call_id,event_type,severity,codec,inbound_frames,inbound_bytes,outbound_frames,outbound_bytes,xoff_count,xon_count,error_redacted,created_at,metadata) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)`, m.CallID, m.EventType, nonEmpty(m.Severity, "info"), m.Codec, m.InboundFrames, m.InboundBytes, m.OutboundFrames, m.OutboundBytes, m.XOffCount, m.XOnCount, m.Error, m.CreatedAt, jsonb(m.Metadata))
|
|
return err
|
|
}
|
|
|
|
func (r *PostgresRepository) ExportCall(ctx context.Context, callID string) (CallAuditExport, error) {
|
|
ex := CallAuditExport{Events: []EventRecord{}, Transcripts: []TranscriptRecord{}, Tools: []ToolAuditRecord{}, KB: []KBAuditRecord{}, Handoffs: []HandoffAuditRecord{}, Providers: []ProviderAuditRecord{}, Media: []MediaAuditRecord{}}
|
|
var c CallRecord
|
|
var meta []byte
|
|
err := r.pool.QueryRow(ctx, `SELECT call_id,coalesce(asterisk_channel_id,''),route,coalesce(caller_number_masked,''),coalesce(language,''),coalesce(region_code,''),coalesce(state,''),started_at,ended_at,coalesce(duration_ms,0),coalesce(end_reason,''),coalesce(handoff_status,''),metadata FROM ai_calls WHERE call_id=$1`, callID).Scan(&c.CallID, &c.AsteriskChannelID, &c.Route, &c.CallerNumberMasked, &c.Language, &c.RegionCode, &c.State, &c.StartedAt, &c.EndedAt, &c.DurationMS, &c.EndReason, &c.HandoffStatus, &meta)
|
|
if err != nil && err != pgx.ErrNoRows {
|
|
return ex, err
|
|
}
|
|
if err == nil {
|
|
c.Metadata = unjson(meta)
|
|
ex.Call = &c
|
|
}
|
|
rows, err := r.pool.Query(ctx, `SELECT event_type,event_source,coalesce(state_before,''),coalesce(state_after,''),severity,coalesce(message,''),coalesce(reason_code,''),created_at,metadata FROM ai_call_events WHERE call_id=$1 ORDER BY created_at`, callID)
|
|
if err != nil {
|
|
return ex, err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var e EventRecord
|
|
var mb []byte
|
|
e.CallID = callID
|
|
if err := rows.Scan(&e.EventType, &e.EventSource, &e.StateBefore, &e.StateAfter, &e.Severity, &e.Message, &e.ReasonCode, &e.CreatedAt, &mb); err != nil {
|
|
return ex, err
|
|
}
|
|
e.Metadata = unjson(mb)
|
|
ex.Events = append(ex.Events, e)
|
|
}
|
|
rows, err = r.pool.Query(ctx, `SELECT speaker,event_type,coalesce(language,''),coalesce(text_redacted,''),created_at,metadata FROM ai_transcript_events WHERE call_id=$1 ORDER BY created_at`, callID)
|
|
if err != nil {
|
|
return ex, err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var t TranscriptRecord
|
|
var mb []byte
|
|
t.CallID = callID
|
|
if err := rows.Scan(&t.Speaker, &t.EventType, &t.Language, &t.Text, &t.CreatedAt, &mb); err != nil {
|
|
return ex, err
|
|
}
|
|
t.Metadata = unjson(mb)
|
|
ex.Transcripts = append(ex.Transcripts, t)
|
|
}
|
|
rows, err = r.pool.Query(ctx, `SELECT coalesce(tool_call_id,''),tool_name,coalesce(state,''),coalesce(language,''),coalesce(region_code,''),allowed,denied,coalesce(reason_code,''),args_redacted,result_redacted,coalesce(duration_ms,0),created_at,metadata FROM ai_tool_audit WHERE call_id=$1 ORDER BY created_at`, callID)
|
|
if err != nil {
|
|
return ex, err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var t ToolAuditRecord
|
|
var ab, rb, mb []byte
|
|
t.CallID = callID
|
|
if err := rows.Scan(&t.ToolCallID, &t.ToolName, &t.State, &t.Language, &t.RegionCode, &t.Allowed, &t.Denied, &t.ReasonCode, &ab, &rb, &t.DurationMS, &t.CreatedAt, &mb); err != nil {
|
|
return ex, err
|
|
}
|
|
t.Args = unjson(ab)
|
|
t.Result = unjson(rb)
|
|
t.Metadata = unjson(mb)
|
|
ex.Tools = append(ex.Tools, t)
|
|
}
|
|
rows, err = r.pool.Query(ctx, `SELECT query_redacted,language,region_code,result_count,coalesce(top_score,0),cross_language_fallback_used,citations_count,no_answer,coalesce(duration_ms,0),created_at,metadata FROM ai_kb_audit WHERE call_id=$1 ORDER BY created_at`, callID)
|
|
if err != nil {
|
|
return ex, err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var k KBAuditRecord
|
|
var mb []byte
|
|
k.CallID = callID
|
|
if err := rows.Scan(&k.Query, &k.Language, &k.RegionCode, &k.ResultCount, &k.TopScore, &k.CrossLanguageFallbackUsed, &k.CitationsCount, &k.NoAnswer, &k.DurationMS, &k.CreatedAt, &mb); err != nil {
|
|
return ex, err
|
|
}
|
|
k.Metadata = unjson(mb)
|
|
ex.KB = append(ex.KB, k)
|
|
}
|
|
rows, err = r.pool.Query(ctx, `SELECT coalesce(handoff_id,''),mode,status,coalesce(reason_code,''),transfer_attempted,transfer_succeeded,coalesce(target_redacted,''),coalesce(summary_redacted,''),created_at,metadata FROM ai_handoff_audit WHERE call_id=$1 ORDER BY created_at`, callID)
|
|
if err != nil {
|
|
return ex, err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var h HandoffAuditRecord
|
|
var mb []byte
|
|
h.CallID = callID
|
|
if err := rows.Scan(&h.HandoffID, &h.Mode, &h.Status, &h.ReasonCode, &h.TransferAttempted, &h.TransferSucceeded, &h.Target, &h.Summary, &h.CreatedAt, &mb); err != nil {
|
|
return ex, err
|
|
}
|
|
h.Metadata = unjson(mb)
|
|
ex.Handoffs = append(ex.Handoffs, h)
|
|
}
|
|
rows, err = r.pool.Query(ctx, `SELECT provider,event_type,severity,coalesce(error_redacted,''),input_audio_bytes,output_audio_bytes,events_received,events_sent,created_at,metadata FROM ai_provider_audit WHERE call_id=$1 ORDER BY created_at`, callID)
|
|
if err != nil {
|
|
return ex, err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var pr ProviderAuditRecord
|
|
var mb []byte
|
|
pr.CallID = callID
|
|
if err := rows.Scan(&pr.Provider, &pr.EventType, &pr.Severity, &pr.Error, &pr.InputAudioBytes, &pr.OutputAudioBytes, &pr.EventsReceived, &pr.EventsSent, &pr.CreatedAt, &mb); err != nil {
|
|
return ex, err
|
|
}
|
|
pr.Metadata = unjson(mb)
|
|
ex.Providers = append(ex.Providers, pr)
|
|
}
|
|
rows, err = r.pool.Query(ctx, `SELECT event_type,severity,coalesce(codec,''),inbound_frames,inbound_bytes,outbound_frames,outbound_bytes,xoff_count,xon_count,coalesce(error_redacted,''),created_at,metadata FROM ai_media_audit WHERE call_id=$1 ORDER BY created_at`, callID)
|
|
if err != nil {
|
|
return ex, err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var m MediaAuditRecord
|
|
var mb []byte
|
|
m.CallID = callID
|
|
if err := rows.Scan(&m.EventType, &m.Severity, &m.Codec, &m.InboundFrames, &m.InboundBytes, &m.OutboundFrames, &m.OutboundBytes, &m.XOffCount, &m.XOnCount, &m.Error, &m.CreatedAt, &mb); err != nil {
|
|
return ex, err
|
|
}
|
|
m.Metadata = unjson(mb)
|
|
ex.Media = append(ex.Media, m)
|
|
}
|
|
return ex, nil
|
|
}
|
|
|
|
func (r *PostgresRepository) Prune(ctx context.Context, req RetentionPruneRequest) (RetentionPruneResult, error) {
|
|
if req.Now.IsZero() {
|
|
req.Now = time.Now().UTC()
|
|
}
|
|
if req.RetentionDays <= 0 {
|
|
req.RetentionDays = 180
|
|
}
|
|
if req.TranscriptRetentionDays <= 0 {
|
|
req.TranscriptRetentionDays = 30
|
|
}
|
|
res := RetentionPruneResult{DryRun: req.DryRun, Status: "ok"}
|
|
_ = r.pool.QueryRow(ctx, `SELECT count(*) FROM ai_calls WHERE started_at < $1`, req.Now.AddDate(0, 0, -req.RetentionDays)).Scan(&res.CallsDeleted)
|
|
_ = r.pool.QueryRow(ctx, `SELECT count(*) FROM ai_call_events WHERE created_at < $1`, req.Now.AddDate(0, 0, -req.RetentionDays)).Scan(&res.EventsDeleted)
|
|
_ = r.pool.QueryRow(ctx, `SELECT count(*) FROM ai_transcript_events WHERE created_at < $1`, req.Now.AddDate(0, 0, -req.TranscriptRetentionDays)).Scan(&res.TranscriptsDeleted)
|
|
_ = r.pool.QueryRow(ctx, `SELECT count(*) FROM ai_tool_audit WHERE created_at < $1`, req.Now.AddDate(0, 0, -req.ToolAuditRetentionDays)).Scan(&res.ToolAuditDeleted)
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO ai_audit_retention_runs(dry_run,status,finished_at,calls_deleted,events_deleted,transcripts_deleted,tool_audit_deleted,metadata) VALUES($1,$2,now(),$3,$4,$5,$6,'{}')`, req.DryRun, res.Status, res.CallsDeleted, res.EventsDeleted, res.TranscriptsDeleted, res.ToolAuditDeleted)
|
|
if err != nil {
|
|
return res, err
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
func jsonb(v map[string]any) []byte {
|
|
if v == nil {
|
|
v = map[string]any{}
|
|
}
|
|
b, _ := json.Marshal(v)
|
|
return b
|
|
}
|
|
func unjson(b []byte) map[string]any {
|
|
out := map[string]any{}
|
|
_ = json.Unmarshal(b, &out)
|
|
return out
|
|
}
|
|
func nonEmpty(v, fallback string) string {
|
|
if v == "" {
|
|
return fallback
|
|
}
|
|
return v
|
|
}
|
|
func wrapErr(op string, err error) error {
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("%s: %w", op, err)
|
|
}
|