package scans import ( "context" "database/sql" "encoding/json" "errors" "sort" "strings" "time" ) // pgText убирает U+0000: PostgreSQL text/json не принимают нулевой байт в UTF-8 (SQLSTATE 22021). func pgText(s string) string { return strings.ReplaceAll(s, "\x00", "") } type PostgresStore struct { db *sql.DB } func NewPostgresStore(db *sql.DB) *PostgresStore { return &PostgresStore{db: db} } func (s *PostgresStore) CreateScan(req CreateScanRequest) (ScanJob, error) { if err := validateCreateScanRequest(&req); err != nil { return ScanJob{}, err } applyDefaultOptions(&req.Options) id, err := newID() if err != nil { return ScanJob{}, err } now := time.Now().UTC() job := ScanJob{ ID: id, Name: req.Name, Status: "queued", CIDRs: req.CIDRs, ExcludeIPs: req.ExcludeIPs, SNMPCredentialsID: req.SNMPCredentialsID, Options: req.Options, Progress: 0, Stats: ScanStats{}, CreatedAt: now, } cidrsJSON, err := json.Marshal(job.CIDRs) if err != nil { return ScanJob{}, err } excludeJSON, err := json.Marshal(job.ExcludeIPs) if err != nil { return ScanJob{}, err } optionsJSON, err := json.Marshal(job.Options) if err != nil { return ScanJob{}, err } statsJSON, err := json.Marshal(job.Stats) if err != nil { return ScanJob{}, err } var snmpID any if v := pgText(job.SNMPCredentialsID); v != "" { snmpID = v } query := ` insert into scan_jobs ( id, name, status, cidrs, exclude_ips, options, progress, stats, snmp_credentials_id, created_at ) values ( $1, $2, $3, $4::jsonb, $5::jsonb, $6::jsonb, $7, $8::jsonb, $9, $10 )` _, err = s.db.ExecContext(context.Background(), query, job.ID, pgText(job.Name), job.Status, string(cidrsJSON), string(excludeJSON), string(optionsJSON), job.Progress, string(statsJSON), snmpID, job.CreatedAt, ) if err != nil { return ScanJob{}, err } return job, nil } func (s *PostgresStore) GetScan(id string) (ScanJob, bool) { query := ` select id, name, status, cidrs::text, exclude_ips::text, options::text, progress, stats::text, coalesce(snmp_credentials_id::text, ''), created_at, started_at, finished_at from scan_jobs where id = $1 limit 1` var job ScanJob var cidrsRaw string var excludeRaw string var optionsRaw string var statsRaw string var startedAt sql.NullTime var finishedAt sql.NullTime err := s.db.QueryRowContext(context.Background(), query, id).Scan( &job.ID, &job.Name, &job.Status, &cidrsRaw, &excludeRaw, &optionsRaw, &job.Progress, &statsRaw, &job.SNMPCredentialsID, &job.CreatedAt, &startedAt, &finishedAt, ) if errors.Is(err, sql.ErrNoRows) { return ScanJob{}, false } if err != nil { return ScanJob{}, false } if err = json.Unmarshal([]byte(cidrsRaw), &job.CIDRs); err != nil { return ScanJob{}, false } if err = json.Unmarshal([]byte(excludeRaw), &job.ExcludeIPs); err != nil { return ScanJob{}, false } if err = json.Unmarshal([]byte(optionsRaw), &job.Options); err != nil { return ScanJob{}, false } if err = json.Unmarshal([]byte(statsRaw), &job.Stats); err != nil { return ScanJob{}, false } if startedAt.Valid { job.StartedAt = startedAt.Time } if finishedAt.Valid { job.FinishedAt = finishedAt.Time } return job, true } func (s *PostgresStore) ListScans(limit int) []ScanJob { if limit <= 0 { limit = 50 } query := ` select id, name, status, cidrs::text, exclude_ips::text, options::text, progress, stats::text, coalesce(snmp_credentials_id::text, ''), created_at, started_at, finished_at from scan_jobs order by created_at desc limit $1` rows, err := s.db.QueryContext(context.Background(), query, limit) if err != nil { return nil } defer rows.Close() jobs := make([]ScanJob, 0, limit) for rows.Next() { var job ScanJob var cidrsRaw string var excludeRaw string var optionsRaw string var statsRaw string var startedAt sql.NullTime var finishedAt sql.NullTime err = rows.Scan( &job.ID, &job.Name, &job.Status, &cidrsRaw, &excludeRaw, &optionsRaw, &job.Progress, &statsRaw, &job.SNMPCredentialsID, &job.CreatedAt, &startedAt, &finishedAt, ) if err != nil { return nil } if json.Unmarshal([]byte(cidrsRaw), &job.CIDRs) != nil || json.Unmarshal([]byte(excludeRaw), &job.ExcludeIPs) != nil || json.Unmarshal([]byte(optionsRaw), &job.Options) != nil || json.Unmarshal([]byte(statsRaw), &job.Stats) != nil { return nil } if startedAt.Valid { job.StartedAt = startedAt.Time } if finishedAt.Valid { job.FinishedAt = finishedAt.Time } jobs = append(jobs, job) } sort.Slice(jobs, func(i, j int) bool { return jobs[i].CreatedAt.After(jobs[j].CreatedAt) }) return jobs } func (s *PostgresStore) UpdateScan(id string, update ScanUpdate) (ScanJob, bool) { current, ok := s.GetScan(id) if !ok { return ScanJob{}, false } if update.Status != "" { current.Status = update.Status } if update.Progress >= 0 { current.Progress = update.Progress } current.Stats = update.Stats if update.StartedAt != nil { current.StartedAt = *update.StartedAt } if update.FinishedAt != nil { current.FinishedAt = *update.FinishedAt } optionsJSON, err := json.Marshal(current.Options) if err != nil { return ScanJob{}, false } statsJSON, err := json.Marshal(current.Stats) if err != nil { return ScanJob{}, false } var startedAt any if !current.StartedAt.IsZero() { startedAt = current.StartedAt } var finishedAt any if !current.FinishedAt.IsZero() { finishedAt = current.FinishedAt } query := ` update scan_jobs set status = $2, options = $3::jsonb, progress = $4, stats = $5::jsonb, started_at = $6, finished_at = $7 where id = $1` _, err = s.db.ExecContext( context.Background(), query, id, current.Status, string(optionsJSON), current.Progress, string(statsJSON), startedAt, finishedAt, ) if err != nil { return ScanJob{}, false } return current, true } func (s *PostgresStore) SaveHostResult(scanID string, host HostResult) error { query := ` insert into scan_hosts (scan_id, ip, is_up, checked_at) values ($1, $2, $3, $4)` _, err := s.db.ExecContext(context.Background(), query, scanID, host.IP, host.IsUp, host.CheckedAt) return err } func (s *PostgresStore) ListHostResults(scanID string, limit int) []HostResult { if limit <= 0 { limit = 200000 } // Одна актуальная строка на IP (иначе при limit по времени часть хостов без SNMP в UI). query := ` select ip::text, is_up, checked_at from ( select distinct on (ip) ip, is_up, checked_at from scan_hosts where scan_id = $1 order by ip asc, checked_at desc ) t order by checked_at desc limit $2` rows, err := s.db.QueryContext(context.Background(), query, scanID, limit) if err != nil { return nil } defer rows.Close() out := make([]HostResult, 0, limit) for rows.Next() { var h HostResult if err = rows.Scan(&h.IP, &h.IsUp, &h.CheckedAt); err != nil { return nil } out = append(out, h) } return out } func (s *PostgresStore) SaveOpenPortResult(scanID string, result OpenPortResult) error { query := ` insert into scan_ports (scan_id, ip, port, is_open, checked_at) values ($1, $2, $3, $4, $5)` _, err := s.db.ExecContext( context.Background(), query, scanID, result.IP, result.Port, result.IsOpen, result.CheckedAt, ) return err } func (s *PostgresStore) ListOpenPortResults(scanID string, limit int) []OpenPortResult { if limit <= 0 { limit = 1000 } query := ` select ip::text, port, is_open, checked_at from scan_ports where scan_id = $1 order by checked_at desc limit $2` rows, err := s.db.QueryContext(context.Background(), query, scanID, limit) if err != nil { return nil } defer rows.Close() out := make([]OpenPortResult, 0, limit) for rows.Next() { var p OpenPortResult if err = rows.Scan(&p.IP, &p.Port, &p.IsOpen, &p.CheckedAt); err != nil { return nil } out = append(out, p) } return out } func (s *PostgresStore) SaveSNMPResult(scanID string, result SNMPResult) error { query := ` insert into scan_snmp (scan_id, ip, success, sys_name, sys_descr, sys_object_id, error_text, checked_at) values ($1, $2, $3, $4, $5, $6, $7, $8)` _, err := s.db.ExecContext( context.Background(), query, scanID, result.IP, result.Success, pgText(result.SysName), pgText(result.SysDescr), pgText(result.SysObjectID), pgText(result.Error), result.CheckedAt, ) return err } func (s *PostgresStore) ListSNMPResults(scanID string, limit int) []SNMPResult { if limit <= 0 { limit = 200000 } query := ` select ip::text, success, coalesce(sys_name, ''), coalesce(sys_descr, ''), coalesce(sys_object_id, ''), coalesce(error_text, ''), checked_at from ( select distinct on (ip) ip, success, sys_name, sys_descr, sys_object_id, error_text, checked_at from scan_snmp where scan_id = $1 order by ip asc, checked_at desc ) t order by checked_at desc limit $2` rows, err := s.db.QueryContext(context.Background(), query, scanID, limit) if err != nil { return nil } defer rows.Close() out := make([]SNMPResult, 0, limit) for rows.Next() { var sres SNMPResult if err = rows.Scan( &sres.IP, &sres.Success, &sres.SysName, &sres.SysDescr, &sres.SysObjectID, &sres.Error, &sres.CheckedAt, ); err != nil { return nil } out = append(out, sres) } return out } func (s *PostgresStore) SaveLLDPResult(scanID string, result LLDPResult) error { query := ` insert into scan_lldp (scan_id, ip, local_port_num, remote_chassis_id, remote_port_id, remote_sys_name, checked_at) values ($1, $2, $3, $4, $5, $6, $7)` _, err := s.db.ExecContext( context.Background(), query, scanID, result.IP, pgText(result.LocalPortNum), pgText(result.RemoteChassisID), pgText(result.RemotePortID), pgText(result.RemoteSysName), result.CheckedAt, ) return err } func (s *PostgresStore) ListLLDPResults(scanID string, limit int) []LLDPResult { if limit <= 0 { limit = 2000 } query := ` select ip::text, coalesce(local_port_num, ''), coalesce(remote_chassis_id, ''), coalesce(remote_port_id, ''), coalesce(remote_sys_name, ''), checked_at from scan_lldp where scan_id = $1 order by checked_at desc limit $2` rows, err := s.db.QueryContext(context.Background(), query, scanID, limit) if err != nil { return nil } defer rows.Close() out := make([]LLDPResult, 0, limit) for rows.Next() { var lr LLDPResult if err = rows.Scan( &lr.IP, &lr.LocalPortNum, &lr.RemoteChassisID, &lr.RemotePortID, &lr.RemoteSysName, &lr.CheckedAt, ); err != nil { return nil } out = append(out, lr) } return out } func (s *PostgresStore) SaveInterfaceResult(scanID string, result InterfaceResult) error { query := ` insert into scan_interfaces (scan_id, ip, if_index, if_descr, if_name, if_oper_status, checked_at) values ($1, $2, $3, $4, $5, $6, $7)` _, err := s.db.ExecContext( context.Background(), query, scanID, result.IP, result.IfIndex, pgText(result.IfDescr), pgText(result.IfName), pgText(result.IfOperStatus), result.CheckedAt, ) return err } func (s *PostgresStore) ListInterfaceResults(scanID string, filterIP string, limit int) []InterfaceResult { if limit <= 0 { limit = 4096 } var rows *sql.Rows var err error if filterIP != "" { rows, err = s.db.QueryContext(context.Background(), ` select ip::text, if_index, coalesce(if_descr, ''), coalesce(if_name, ''), coalesce(if_oper_status, ''), checked_at from scan_interfaces where scan_id = $1 and ip = $2::inet order by if_index asc limit $3`, scanID, filterIP, limit) } else { rows, err = s.db.QueryContext(context.Background(), ` select ip::text, if_index, coalesce(if_descr, ''), coalesce(if_name, ''), coalesce(if_oper_status, ''), checked_at from scan_interfaces where scan_id = $1 order by ip asc, if_index asc limit $2`, scanID, limit) } if err != nil { return nil } defer rows.Close() out := make([]InterfaceResult, 0, 256) for rows.Next() { var r InterfaceResult if err = rows.Scan(&r.IP, &r.IfIndex, &r.IfDescr, &r.IfName, &r.IfOperStatus, &r.CheckedAt); err != nil { return nil } out = append(out, r) } return out } func (s *PostgresStore) ListPortDevicesByMac(scanID string, mac string, limit int) []PortDeviceResult { if limit <= 0 { limit = 5000 } if mac == "" { return nil } query := ` select ip::text, if_index, bridge_port, vlan, coalesce(mac, ''), coalesce(learned_ip::text, ''), checked_at from scan_port_devices where scan_id = $1 and lower(replace(trim(coalesce(mac, '')), '-', ':')) = lower(replace(trim($2::text), '-', ':')) order by ip::text asc, if_index asc, vlan asc, mac asc, learned_ip asc limit $3` rows, err := s.db.QueryContext(context.Background(), query, scanID, mac, limit) if err != nil { return nil } defer rows.Close() out := make([]PortDeviceResult, 0, 32) for rows.Next() { var r PortDeviceResult if err = rows.Scan(&r.IP, &r.IfIndex, &r.BridgePort, &r.Vlan, &r.MAC, &r.LearnedIP, &r.CheckedAt); err != nil { return nil } out = append(out, r) } return out } func (s *PostgresStore) PurgeAllScanData() error { // Дочерние таблицы ссылаются на scan_jobs — CASCADE очищает всё; RESTART IDENTITY сбрасывает bigserial. _, err := s.db.ExecContext(context.Background(), `truncate table scan_jobs restart identity cascade`) return err } func (s *PostgresStore) SavePortDeviceResult(scanID string, result PortDeviceResult) error { query := ` insert into scan_port_devices (scan_id, ip, if_index, bridge_port, vlan, mac, learned_ip, checked_at) values ($1, $2, $3, $4, $5, $6, $7, $8)` var learnedIP any if v := strings.TrimSpace(result.LearnedIP); v != "" { learnedIP = v } else { learnedIP = nil } _, err := s.db.ExecContext( context.Background(), query, scanID, result.IP, result.IfIndex, result.BridgePort, result.Vlan, pgText(result.MAC), learnedIP, result.CheckedAt, ) return err } func (s *PostgresStore) ListPortDeviceResults(scanID string, filterIP string, filterIfIndex int, limit int) []PortDeviceResult { if limit <= 0 { limit = 20000 } query := ` select ip::text, if_index, bridge_port, vlan, coalesce(mac, ''), coalesce(learned_ip::text, ''), checked_at from scan_port_devices where scan_id = $1 and ($2 = '' or ip = $2::inet) and ($3 = 0 or if_index = $3 or bridge_port = $3) order by if_index asc, vlan asc, mac asc, learned_ip asc limit $4` rows, err := s.db.QueryContext(context.Background(), query, scanID, filterIP, filterIfIndex, limit) if err != nil { return nil } defer rows.Close() out := make([]PortDeviceResult, 0, 256) for rows.Next() { var r PortDeviceResult if err = rows.Scan(&r.IP, &r.IfIndex, &r.BridgePort, &r.Vlan, &r.MAC, &r.LearnedIP, &r.CheckedAt); err != nil { return nil } out = append(out, r) } return out }