cfdns/pkg/cf/cf.go

package cf

import (
	"context"
	"net/http"
	"sync"
	"time"

	"github.com/cloudflare/cloudflare-go"
	"github.com/goodieshq/cfdns/pkg/config"
	"github.com/goodieshq/cfdns/pkg/ipget"
	"github.com/goodieshq/goropo"
	"github.com/rs/zerolog"
	"github.com/rs/zerolog/log"
)

const (
	RECORD_TYPE_IPV4 = "A"
	RECORD_TYPE_IPV6 = "AAAA"
)

type CFDNS struct {
	mu         sync.RWMutex    // protects config, api, httpClient, pool
	cfg        config.Config   // current configuration
	api        *cloudflare.API // Cloudflare API client
	httpClient http.Client     // shared HTTP client
	timeout    time.Duration   // HTTP timeout duration
	pool       *goropo.Pool    // worker pool for concurrent tasks
}

// NewCFDNS creates a new Cloudflare DNS updater instance
func NewCFDNS(cfg config.Config) (*CFDNS, error) {
	cfdns := &CFDNS{}
	cfdns.SetConfig(&cfg)
	return cfdns, nil
}

// Close aborts all pending tasks and closes the worker pool
func (cfdns *CFDNS) Close() {
	cfdns.mu.Lock()
	pool := cfdns.pool
	if pool == nil {
		cfdns.mu.Unlock()
		return
	}

	// close while holding write lock so SetConfig/other writers can't race
	pool.Abort()
	cfdns.mu.Unlock()
}

// Wait will wait for all tasks in the pool to complete
func (cfdns *CFDNS) Wait() {
	cfdns.mu.RLock()
	pool := cfdns.pool
	if pool == nil {
		cfdns.mu.RUnlock()
		return
	}

	// keep RLock held while waiting so a concurrent SetConfig (writer) will block
	pool.WaitIdle()
	cfdns.mu.RUnlock()
}

func (cfdns *CFDNS) SetConfig(cfg *config.Config) error {
	if cfg != nil {
		cfdns.mu.Lock()
		defer cfdns.mu.Unlock()

		// create a new cloudflare API client from the scoped token
		api, err := cloudflare.NewWithAPIToken(cfg.Token)
		if err != nil {
			return err
		}

		// swap in the new config and resources
		cfdns.api = api
		cfdns.httpClient = http.Client{
			Timeout: cfg.Timeout,
		}
		if cfdns.pool != nil {
			// close previous pool before replacing
			cfdns.pool.Close()
		}

		cfdns.timeout = cfg.Timeout
		cfdns.pool = goropo.NewPool(cfg.WorkerCount, cfg.WorkerCount*5)

		if cfg.Verbose {
			zerolog.SetGlobalLevel(zerolog.DebugLevel)
		} else {
			zerolog.SetGlobalLevel(zerolog.InfoLevel)
		}
		cfdns.cfg = *cfg
	}
	return nil
}

// getRecords retrieves DNS records for the given hostname and record type
func (cfdns *CFDNS) getRecords(ctx context.Context, hostname, recordType string) ([]cloudflare.DNSRecord, error) {
	cfdns.mu.RLock()
	defer cfdns.mu.RUnlock()
	records, _, err := cfdns.api.ListDNSRecords(
		ctx,
		cloudflare.ZoneIdentifier(cfdns.cfg.ZoneID),
		cloudflare.ListDNSRecordsParams{
			Name: hostname,
			Type: recordType,
		},
	)
	if err != nil {
		return nil, err
	}
	return records, nil
}

// ZoneIsValid checks if the configured zone ID is valid for the provided API token
func (cfdns *CFDNS) ZoneIsValid(ctx context.Context) (bool, error) {
	cfdns.mu.RLock()
	pool := cfdns.pool
	api := cfdns.api
	zoneID := cfdns.cfg.ZoneID

	fut := goropo.Submit(
		pool,
		ctx,
		func(ctx context.Context) ([]cloudflare.Zone, error) {
			return api.ListZones(ctx)
		},
	)

	zones, err := fut.Await(ctx)
	cfdns.mu.RUnlock()

	if err != nil {
		return false, err
	}

	for _, zone := range zones {
		if zone.ID == zoneID {
			return true, nil
		}
	}

	return false, nil
}

// checkAndUpdate checks the existing DNS records for the given domain and record type,
// and updates or creates the record if the address has changed or does not exist.
// Caller must hold cfdns.mu RLock.
func (cfdns *CFDNS) checkAndUpdate(
	ctx context.Context,
	domain *config.Domain,
	recordType,
	address string,
) error {
	const timeout = time.Second * 10

	// get existing records for this hostname and record type
	ctxTimeout, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	records, err := cfdns.getRecords(ctxTimeout, domain.Hostname, recordType)
	if err != nil {
		return err
	}

	// no records found for this hostname, create a new one
	if len(records) == 0 {
		ctxTimeout, cancel = context.WithTimeout(ctx, timeout)
		defer cancel()

		recordNew, err := cfdns.api.CreateDNSRecord(
			ctxTimeout,
			cloudflare.ZoneIdentifier(cfdns.cfg.ZoneID),
			cloudflare.CreateDNSRecordParams{
				Name:    domain.Hostname,
				Content: address,
				Type:    recordType,
				Proxied: domain.Proxied,
			},
		)
		if err != nil {
			return err
		}
		log.Info().
			Str("id", recordNew.ID).
			Str("hostname", recordNew.Name).
			Str("type", recordNew.Type).
			Str("address", recordNew.Content).
			Msgf("Created new DNS record")
		return nil
	}

	// iterate over existing records and update if the address has changed
	for _, record := range records {
		ctxTimeout, cancel = context.WithTimeout(ctx, timeout)
		defer cancel()

		if record.Content == address && (record.Proxied == nil ||
			domain.Proxied == nil ||
			(*record.Proxied == *domain.Proxied)) {

			log.Debug().
				Str("id", record.ID).
				Str("hostname", record.Name).
				Str("type", record.Type).
				Str("address", record.Content).
				Msgf("Skipping DNS record")
			return nil
		}

		recordNew, err := cfdns.api.UpdateDNSRecord(
			ctxTimeout,
			cloudflare.ZoneIdentifier(cfdns.cfg.ZoneID),
			cloudflare.UpdateDNSRecordParams{
				ID:      record.ID,
				Content: address,
				Proxied: domain.Proxied,
			},
		)
		if err != nil {
			log.Error().Err(err).
				Str("id", record.ID).
				Str("hostname", record.Name).
				Str("type", record.Type).
				Str("address", record.Content).
				Msg("Failed to update DNS record")
			return err
		}

		log.Info().
			Str("id", recordNew.ID).
			Str("hostname", recordNew.Name).
			Str("type", recordNew.Type).
			Str("address", recordNew.Content).
			Msgf("Updated DNS record")
	}
	return nil
}

func (cfdns *CFDNS) getPublicIPs(ctx context.Context) (string, string) {
	var fut4, fut6 *goropo.Future[string]
	var ipv4, ipv6 string

	if *cfdns.cfg.IPv4 {
		fut4 = goropo.Submit(cfdns.pool, ctx, ipget.GetPublicIPv4)
	}

	if *cfdns.cfg.IPv6 {
		fut6 = goropo.Submit(cfdns.pool, ctx, ipget.GetPublicIPv6)
	}

	if fut4 != nil {
		v, err := fut4.Await(ctx)
		if err != nil {
			log.Error().Err(err).Msg("failed to get public ipv4")
			ipv4 = ""
		} else {
			ipv4 = v
			log.Debug().Str("ipv4", ipv4).Msg("fetched ipv4 address")
		}
	}

	if fut6 != nil {
		v, err := fut6.Await(ctx)
		if err != nil {
			log.Error().Err(err).Msg("failed to get public ipv6")
			ipv6 = ""
		} else {
			ipv6 = v
			log.Debug().Str("ipv6", ipv6).Msg("fetched ipv6 address")
		}
	}

	return ipv4, ipv6
}

func (cfdns *CFDNS) Process(ctx context.Context) {
	cfdns.mu.RLock()
	defer cfdns.mu.RUnlock()

	valid, err := cfdns.ZoneIsValid(ctx)
	if err != nil || !valid {
		log.Error().Err(err).Msg("unable to verify zone and API token, skipping processing cycle")
		return
	}

	// acquire the current public IP addresses for this run
	ipv4, ipv6 := cfdns.getPublicIPs(ctx)

	// make a list of futures for all domain updates, allocate enough for both ipv4 and ipv6
	futs := make([]*goropo.FutureAny, 0, len(cfdns.cfg.Domains)*2)

	// iterate over all configured domains and update their DNS records as needed
	for _, domain := range cfdns.cfg.Domains {
		if *cfdns.cfg.IPv4 && ipv4 != "" {
			fut := goropo.Submit(
				cfdns.pool,
				ctx,
				func(ctx context.Context) (any, error) {
					if err := cfdns.checkAndUpdate(ctx, &domain, RECORD_TYPE_IPV4, ipv4); err != nil {
						log.Error().Err(err).Str("domain", domain.Hostname).Msg("failed to update ipv4 record")
						return nil, err
					}
					return nil, nil
				},
			)
			futs = append(futs, fut)
		}

		if *cfdns.cfg.IPv6 && ipv6 != "" {
			fut := goropo.Submit(
				cfdns.pool,
				ctx,
				func(ctx context.Context) (any, error) {
					if err := cfdns.checkAndUpdate(ctx, &domain, RECORD_TYPE_IPV6, ipv6); err != nil {
						log.Error().Err(err).Str("domain", domain.Hostname).Msg("failed to update ipv6 record")
						return nil, err
					}
					return nil, nil
				},
			)
			futs = append(futs, fut)
		}
	}

	for _, fut := range futs {
		_, _ = fut.Await(ctx)
	}
}