diff --git a/net/iplist/source.go b/net/iplist/source.go new file mode 100644 index 0000000..6027dd7 --- /dev/null +++ b/net/iplist/source.go @@ -0,0 +1,143 @@ +package iplist + +import ( + "context" + "errors" + "net/http" + "net/url" + "os" + "sync" + "sync/atomic" + "time" +) + +const DefaultRefreshInterval = time.Hour + 57*time.Minute + 13*time.Second + +type SourceConfig struct { + Source string + CacheDir string + HTTPClient *http.Client + RefreshInterval time.Duration + Optional bool + OnRefresh func(SourceEvent) +} + +type SourceEvent struct { + Entries uint64 + Err error +} + +type sourceSnapshot struct { + entries []string +} + +type Source struct { + ctx context.Context + cancel context.CancelFunc + config SourceConfig + current atomic.Pointer[sourceSnapshot] + observers []func(SourceEvent) + observerMu sync.RWMutex +} + +func NewSource(ctx context.Context, config SourceConfig) (*Source, error) { + if config.RefreshInterval <= 0 { + config.RefreshInterval = DefaultRefreshInterval + } + sourceCtx, cancel := context.WithCancel(ctx) + s := &Source{ctx: sourceCtx, cancel: cancel, config: config} + entries, err := s.load() + if err != nil { + cancel() + return nil, err + } + s.store(entries) + go s.refreshLoop() + return s, nil +} + +func (s *Source) Entries() []string { + current := s.current.Load() + if current == nil { + return nil + } + return append([]string(nil), current.entries...) +} + +func (s *Source) Refresh() error { + entries, err := s.load() + if err != nil { + s.emit(SourceEvent{Entries: uint64(len(s.Entries())), Err: err}) + return err + } + s.store(entries) + s.emit(SourceEvent{Entries: uint64(len(entries))}) + return nil +} + +func (s *Source) Subscribe(observer func(SourceEvent)) { + if observer == nil { + return + } + s.observerMu.Lock() + s.observers = append(s.observers, observer) + s.observerMu.Unlock() +} + +func (s *Source) emit(event SourceEvent) { + if s.config.OnRefresh != nil { + s.config.OnRefresh(event) + } + s.observerMu.RLock() + observers := append([]func(SourceEvent){}, s.observers...) + s.observerMu.RUnlock() + for _, observer := range observers { + observer(event) + } +} + +func (s *Source) Close() error { + if s != nil && s.cancel != nil { + s.cancel() + } + return nil +} + +func (s *Source) load() ([]string, error) { + if s.config.Source == "" { + if s.config.Optional { + return nil, nil + } + return nil, errors.New("source is required") + } + if s.config.Optional { + parsed, parseErr := url.Parse(s.config.Source) + isURL := parseErr == nil && (parsed.Scheme == "http" || parsed.Scheme == "https") + if !isURL { + if _, err := os.Stat(s.config.Source); os.IsNotExist(err) { + return nil, nil + } else if err != nil { + return nil, err + } + } + } + return Load(s.ctx, s.config.Source, s.config.CacheDir, s.config.HTTPClient) +} + +func (s *Source) store(entries []string) { + copyEntries := append([]string(nil), entries...) + s.current.Store(&sourceSnapshot{entries: copyEntries}) +} + +func (s *Source) refreshLoop() { + ticker := time.NewTicker(s.config.RefreshInterval) + defer ticker.Stop() + for { + select { + case <-s.ctx.Done(): + return + case <-ticker.C: + _ = s.Refresh() + } + } +} diff --git a/net/iplist/source_test.go b/net/iplist/source_test.go new file mode 100644 index 0000000..5e41899 --- /dev/null +++ b/net/iplist/source_test.go @@ -0,0 +1,46 @@ +package iplist + +import ( + "os" + "path/filepath" + "testing" +) + +func TestSourceKeepsLastGoodEntriesAfterFailedRefresh(t *testing.T) { + path := filepath.Join(t.TempDir(), "list.tsv") + if err := os.WriteFile(path, []byte("192.0.2.1\n"), 0o600); err != nil { + t.Fatal(err) + } + source, err := NewSource(t.Context(), SourceConfig{Source: path}) + if err != nil { + t.Fatal(err) + } + defer func() { _ = source.Close() }() + + if got := source.Entries(); len(got) != 1 || got[0] != "192.0.2.1" { + t.Fatalf("initial entries = %#v", got) + } + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + if err := source.Refresh(); err == nil { + t.Fatal("invalid refresh unexpectedly succeeded") + } + if got := source.Entries(); len(got) != 1 || got[0] != "192.0.2.1" { + t.Fatalf("entries after failed refresh = %#v", got) + } +} + +func TestSourceOptionalMissingFileIsEmpty(t *testing.T) { + source, err := NewSource(t.Context(), SourceConfig{ + Source: filepath.Join(t.TempDir(), "missing.tsv"), + Optional: true, + }) + if err != nil { + t.Fatal(err) + } + defer func() { _ = source.Close() }() + if got := source.Entries(); len(got) != 0 { + t.Fatalf("entries = %#v", got) + } +}