depot/go/trains/darwin/darwingest/darwingestftp/darwingestftp.go

package darwingestftp

import (
	"compress/gzip"
	"context"
	"encoding/xml"
	"errors"
	"fmt"
	"io"
	"log"
	"sort"
	"time"

	"github.com/jlaffaye/ftp"
	"hg.lukegb.com/lukegb/depot/go/trains/darwin"
)

func loadGzippedFile(ctx context.Context, sc *ftp.ServerConn, fn string, cb func(pp *darwin.PushPort) error) error {
	log.Printf("fetching %v", fn)
	resp, err := sc.Retr(fn)
	if err != nil {
		return fmt.Errorf("Retr(%q): %w", fn, err)
	}
	defer resp.Close()

	respGz, err := gzip.NewReader(resp)
	if err != nil {
		return fmt.Errorf("gzip.NewReader from %q: %w", fn, err)
	}
	defer respGz.Close()

	xmlDec := xml.NewDecoder(respGz)
	for {
		if ctx.Err() != nil {
			return ctx.Err()
		}

		var pp darwin.PushPort
		if err := xmlDec.Decode(&pp); errors.Is(err, io.EOF) {
			break
		} else if err != nil {
			return fmt.Errorf("xmlDec.Decode from %q: %w", fn, err)
		}

		if err := cb(&pp); err != nil {
			return fmt.Errorf("processing %q: %w", fn, err)
		}
	}
	return nil
}

func LoadSnapshot(ctx context.Context, sc *ftp.ServerConn, cb func(*darwin.PushPort) error) (lowWater, highWater time.Time, err error) {
	log.Println("fetching snapshot...")
	err = loadGzippedFile(ctx, sc, "/snapshot/snapshot.gz", func(pp *darwin.PushPort) error {
		if lowWater.IsZero() || lowWater.After(pp.Time) {
			lowWater = pp.Time
		}
		if highWater.IsZero() || highWater.Before(pp.Time) {
			highWater = pp.Time
		}
		return cb(pp)
	})
	if err != nil {
		return time.Time{}, time.Time{}, err
	}
	return lowWater, highWater, nil
}

type logFTPFile struct {
	Filename string
	EndAt    time.Time
}

func LoadLogs(ctx context.Context, sc *ftp.ServerConn, startAt, endAt time.Time, cb func(*darwin.PushPort) error) error {
	firstIt := true
	for {
		if ctx.Err() != nil {
			return ctx.Err()
		}

		log.Println("fetching logs starting at", startAt)
		entries, err := sc.List("/pushport")
		if err != nil {
			return fmt.Errorf("listing /pushport: %w", err)
		}

		var fs []logFTPFile
		for _, e := range entries {
			if e.Type != ftp.EntryTypeFile {
				continue
			}

			t, err := time.ParseInLocation("pPortData.log.2006-01-02_1504.gz", e.Name, darwin.London)
			if err != nil {
				log.Println("failed to parse filename %q: %v", err)
				continue
			}

			fs = append(fs, logFTPFile{
				Filename: e.Name,
				EndAt:    t,
			})
		}

		sort.Slice(fs, func(i, j int) bool {
			return fs[i].EndAt.Before(fs[j].EndAt)
		})

		startReadingFrom := -1
		endReadingAt := len(fs) - 1
		for n, f := range fs {
			if f.EndAt.After(startAt) && startReadingFrom == -1 {
				startReadingFrom = n
			}
			if f.EndAt.After(endAt) && !endAt.IsZero() {
				endReadingAt = n
				break
			}
		}

		var lowWater, highWater time.Time
		if startReadingFrom != -1 {
			for n := startReadingFrom; n <= endReadingAt; n++ {
				fmt.Printf("%v: ", fs[n].Filename)
				err := loadGzippedFile(ctx, sc, fmt.Sprintf("/pushport/%v", fs[n].Filename), func(pp *darwin.PushPort) error {
					if lowWater.IsZero() || lowWater.After(pp.Time) {
						lowWater = pp.Time
					}
					if highWater.IsZero() || highWater.Before(pp.Time) {
						highWater = pp.Time
					}
					if !pp.Time.After(startAt) {
						fmt.Printf("x")
						return nil
					} else if !endAt.IsZero() && (endAt.Before(pp.Time) || endAt.Equal(pp.Time)) {
						fmt.Printf("_")
						return nil
					}
					return cb(pp)
				})
				if err != nil {
					return err
				}
				fmt.Print("\n")
			}

			if !lowWater.Before(startAt) {
				// Don't believe that we saw everything :/
				log.Printf("low water was %v; start at was %v; we might not have seen everything!", lowWater, startAt)
			}
		}

		if !highWater.After(endAt) {
			if !firstIt {
				log.Printf("not yet caught up: high water %v; end at %v; waiting 1 minute...", highWater, endAt)
				if err := darwin.Sleep(ctx, 1*time.Minute); err != nil {
					return err
				}
			}
			firstIt = false
			if !highWater.IsZero() {
				startAt = highWater
			}
			continue
		}

		return nil
	}
}