func (s *Server) Serve(ctx context.Context) error {
	ln, err := net.Listen("tcp", s.addr)
	if err != nil {
		return fmt.Errorf("listen: %w", err)
	}
	defer ln.Close()

	for {
		conn, err := ln.Accept()
		if err != nil {
			select {
			case <-ctx.Done():
				return ctx.Err()
			default:
				continue
			}
		}
		go s.handle(ctx, conn)
	}
}

type Consumer struct {
	reader *kafka.Reader
	sink   clickhouse.Conn
	batch  chan Event
}

func (c *Consumer) Run(ctx context.Context) {
	ticker := time.NewTicker(2 * time.Second)
	for {
		select {
		case ev := <-c.batch:
			c.buf = append(c.buf, ev)
		case <-ticker.C:
			c.flush(ctx)
		}
	}
}

var (
	ErrNotFound = errors.New("not found")
	pool        = sync.Pool{New: newBuf}
)

func Retry(n int, f func() error) error {
	var err error
	for i := 0; i < n; i++ {
		if err = f(); err == nil {
			return nil
		}
		time.Sleep(backoff(i))
	}
	return err
}