// addrToHostPort takes the given address and parses it into a string suitable
// for use in the 'hostnames' field in a known_hosts file.  For more details,
// see the `SSH_KNOWN_HOSTS FILE FORMAT` section of `man 8 sshd`
func (kc *HostKeyChecker) addrToHostPort(a string) (string, error) {
	if !strings.Contains(a, ":") {
		// No port, so return unadulterated
		return a, nil
	}
	host, p, err := net.SplitHostPort(a)
	if err != nil {
		log.Debugf("Unable to parse addr %s: %v", a, err)
		return "", err
	}

	port, err := strconv.Atoi(p)
	if err != nil {
		log.Debugf("Error parsing port %s: %v", p, err)
		return "", err
	}

	// Default port should be omitted from the entry.
	// (see `put_host_port` in openssh/misc.c)
	if port == 0 || port == sshDefaultPort {
		// IPv6 addresses must be enclosed in square brackets
		if strings.Contains(host, ":") {
			host = fmt.Sprintf("[%s]", host)
		}
		return host, nil
	}

	return fmt.Sprintf("[%s]:%d", host, port), nil
}

func watch(kAPI etcd.KeysAPI, key string, stop chan struct{}) (res *etcd.Response) {
	for res == nil {
		select {
		case <-stop:
			log.Debugf("Gracefully closing etcd watch loop: key=%s", key)
			return
		default:
			opts := &etcd.WatcherOptions{
				AfterIndex: 0,
				Recursive:  true,
			}
			watcher := kAPI.Watcher(key, opts)
			log.Debugf("Creating etcd watcher: %s", key)

			var err error
			res, err = watcher.Next(context.Background())
			if err != nil {
				log.Errorf("etcd watcher %v returned error: %v", key, err)
			}
		}

		// Let's not slam the etcd server in the event that we know
		// an unexpected error occurred.
		time.Sleep(time.Second)
	}

	return
}

// newPublisher returns a publishFunc that publishes a single UnitState
// by the given name to the provided Registry, with the given TTL
func newPublisher(reg registry.Registry, ttl time.Duration) publishFunc {
	return func(name string, us *unit.UnitState) {
		if us == nil {
			log.Debugf("Destroying UnitState(%s) in Registry", name)
			err := reg.RemoveUnitState(name)
			if err != nil {
				log.Errorf("Failed to destroy UnitState(%s) in Registry: %v", name, err)
			}
		} else {
			// Sanity check - don't want to publish incomplete UnitStates
			// TODO(jonboulle): consider teasing apart a separate UnitState-like struct
			// so we can rely on a UnitState always being fully hydrated?

			// See https://github.com/coreos/fleet/issues/720
			//if len(us.UnitHash) == 0 {
			//	log.Errorf("Refusing to push UnitState(%s), no UnitHash: %#v", name, us)

			if len(us.MachineID) == 0 {
				log.Errorf("Refusing to push UnitState(%s), no MachineID: %#v", name, us)
			} else {
				log.Debugf("Pushing UnitState(%s) to Registry: %#v", name, us)
				reg.SaveUnitState(name, us, ttl)
			}
		}
	}
}

func getDefaultGatewayIface() *net.Interface {
	log.Debug("Attempting to retrieve IP route info from netlink")

	routes, err := netlink.RouteList(nil, 0)
	if err != nil {
		log.Debugf("Unable to detect default interface: %v", err)
		return nil
	}

	if len(routes) == 0 {
		log.Debugf("Netlink returned zero routes")
		return nil
	}

	for _, route := range routes {
		// a nil Dst means that this is the default route.
		if route.Dst == nil {
			i, err := net.InterfaceByIndex(route.LinkIndex)
			if err != nil {
				log.Debugf("Found default route but could not determine interface")
				continue
			}
			log.Debugf("Found default route with interface %v", i)
			return i
		}
	}

	log.Debugf("Unable to find default route")
	return nil
}

func runUnloadUnit(args []string) (exit int) {
	if len(args) == 0 {
		stderr("No units given")
		return 0
	}

	units, err := findUnits(args)
	if err != nil {
		stderr("%v", err)
		return 1
	}

	wait := make([]string, 0)
	for _, s := range units {
		if !suToGlobal(s) {
			if job.JobState(s.CurrentState) == job.JobStateInactive {
				log.Debugf("Target state of Unit(%s) already %s, skipping.", s.Name, job.JobStateInactive)
				continue
			}
		}

		log.Debugf("Setting target state of Unit(%s) to %s", s.Name, job.JobStateInactive)
		cAPI.SetUnitTargetState(s.Name, string(job.JobStateInactive))
		if suToGlobal(s) {
			stdout("Triggered global unit %s unload", s.Name)
		} else {
			wait = append(wait, s.Name)
		}
	}

	exit = tryWaitForUnitStates(wait, "unload", job.JobStateInactive, getBlockAttempts(), os.Stdout)

	return
}

// ParseFilepath expands ~ and ~user constructions.
// If user or $HOME is unknown, do nothing.
func ParseFilepath(path string) string {
	if !strings.HasPrefix(path, "~") {
		return path
	}
	i := strings.Index(path, "/")
	if i < 0 {
		i = len(path)
	}
	var home string
	if i == 1 {
		if home = os.Getenv("HOME"); home == "" {
			usr, err := user.Current()
			if err != nil {
				log.Debugf("Failed to get current home directory: %v", err)
				return path
			}
			home = usr.HomeDir
		}
	} else {
		usr, err := user.Lookup(path[1:i])
		if err != nil {
			log.Debugf("Failed to get %v's home directory: %v", path[1:i], err)
			return path
		}
		home = usr.HomeDir
	}
	path = filepath.Join(home, path[i:])
	return path
}

func getDefaultGatewayIface() *net.Interface {
	log.Debug("Attempting to retrieve IP route info from netlink")

	routes, err := netlink.NetworkGetRoutes()
	if err != nil {
		log.Debugf("Unable to detect default interface: %v", err)
		return nil
	}

	if len(routes) == 0 {
		log.Debugf("Netlink returned zero routes")
		return nil
	}

	for _, route := range routes {
		if route.Default {
			if route.Iface == nil {
				log.Debugf("Found default route but could not determine interface")
			}
			log.Debugf("Found default route with interface %v", route.Iface.Name)
			return route.Iface
		}
	}

	log.Debugf("Unable to find default route")
	return nil
}

func watch(client etcd.Client, key string, stop chan struct{}) (res *etcd.Result) {
	for res == nil {
		select {
		case <-stop:
			log.Debugf("Gracefully closing etcd watch loop: key=%s", key)
			return
		default:
			req := &etcd.Watch{
				Key:       key,
				WaitIndex: 0,
				Recursive: true,
			}

			log.Debugf("Creating etcd watcher: %v", req)

			var err error
			res, err = client.Wait(req, stop)
			if err != nil {
				log.Errorf("etcd watcher %v returned error: %v", req, err)
			}
		}

		// Let's not slam the etcd server in the event that we know
		// an unexpected error occurred.
		time.Sleep(time.Second)
	}

	return
}

func (lt *LoggingHTTPTransport) RoundTrip(req *http.Request) (resp *http.Response, err error) {
	log.Debugf("HTTP %s %s", req.Method, req.URL.String())
	resp, err = lt.Transport.RoundTrip(req)
	if err == nil {
		log.Debugf("HTTP %s %s %s", req.Method, req.URL.String(), resp.Status)
	}
	return
}

func (ar *actionResolver) one(req *http.Request, cancel <-chan struct{}) (resp *http.Response, body []byte, err error) {
	log.Debugf("etcd: sending HTTP request %s %s", req.Method, req.URL)
	resp, body, err = ar.requestFunc(req, cancel)
	if err != nil {
		log.Debugf("etcd: recv error response from %s %s: %v", req.Method, req.URL, err)
		return
	}

	log.Debugf("etcd: recv response from %s %s: %s", req.Method, req.URL, resp.Status)
	return
}

func (m *systemdUnitManager) GetUnitStates(filter pkg.Set) (map[string]*unit.UnitState, error) {
	// Unfortunately we need to lock for the entire operation to ensure we
	// have a consistent view of the hashes. Otherwise, Load/Unload
	// operations could mutate the hashes before we've retrieved the state
	// for every unit in the filter, since they won't necessarily all be
	// present in the initial ListUnits() call.
	fallback := false

	m.mutex.Lock()
	defer m.mutex.Unlock()
	dbusStatuses, err := m.systemd.ListUnitsByNames(filter.Values())

	if err != nil {
		fallback = true
		log.Debugf("ListUnitsByNames is not implemented in your systemd version (requires at least systemd 230), fallback to ListUnits: %v", err)
		dbusStatuses, err = m.systemd.ListUnits()
		if err != nil {
			return nil, err
		}
	}

	states := make(map[string]*unit.UnitState)
	for _, dus := range dbusStatuses {
		if fallback && !filter.Contains(dus.Name) {
			// If filter could not be applied on DBus side, we will filter unit files here
			continue
		}

		us := &unit.UnitState{
			LoadState:   dus.LoadState,
			ActiveState: dus.ActiveState,
			SubState:    dus.SubState,
		}
		if h, ok := m.hashes[dus.Name]; ok {
			us.UnitHash = h.String()
		}
		states[dus.Name] = us
	}

	// grab data on subscribed units that didn't show up in ListUnits in fallback mode, most
	// likely due to being inactive
	if fallback {
		for _, name := range filter.Values() {
			if _, ok := states[name]; ok {
				continue
			}

			us, err := m.getUnitState(name)
			if err != nil {
				return nil, err
			}
			if h, ok := m.hashes[name]; ok {
				us.UnitHash = h.String()
			}
			states[name] = us
		}
	}

	return states, nil
}

func createUnit(name string, uf *unit.UnitFile) (*schema.Unit, error) {
	if uf == nil {
		return nil, fmt.Errorf("nil unit provided")
	}
	u := schema.Unit{
		Name:    name,
		Options: schema.MapUnitFileToSchemaUnitOptions(uf),
	}
	// TODO(jonboulle): this dependency on the API package is awkward, and
	// redundant with the check in api.unitsResource.set, but it is a
	// workaround to implementing the same check in the RegistryClient. It
	// will disappear once RegistryClient is deprecated.
	if err := api.ValidateName(name); err != nil {
		return nil, err
	}
	if err := api.ValidateOptions(u.Options); err != nil {
		return nil, err
	}
	j := &job.Job{Unit: *uf}
	if err := j.ValidateRequirements(); err != nil {
		log.Warningf("Unit %s: %v", name, err)
	}
	err := cAPI.CreateUnit(&u)
	if err != nil {
		return nil, fmt.Errorf("failed creating unit %s: %v", name, err)
	}

	log.Debugf("Created Unit(%s) in Registry", name)
	return &u, nil
}

// check attempts to beat a Heart several times within a timeout, returning the
// log index at which the beat succeeded or an error
func check(hrt heart.Heart, ttl time.Duration) (idx uint64, err error) {
	// time out after a third of the machine presence TTL, attempting
	// the heartbeat up to four times
	timeout := ttl / 3
	interval := timeout / 4

	tchan := time.After(timeout)
	next := time.After(0)
	for idx == 0 {
		select {
		case <-tchan:
			err = errors.New("Monitor timed out before successful heartbeat")
			return
		case <-next:
			idx, err = hrt.Beat(ttl)
			if err != nil {
				log.Debugf("Monitor heartbeat function returned err, retrying in %v: %v", interval, err)
			}

			next = time.After(interval)
		}
	}

	return
}

func assertUnitState(name string, js job.JobState, out io.Writer) (ret bool) {
	u, err := cAPI.Unit(name)
	if err != nil {
		log.Warningf("Error retrieving Unit(%s) from Registry: %v", name, err)
		return
	}
	if u == nil {
		log.Warningf("Unit %s not found", name)
		return
	}
	if job.JobState(u.CurrentState) != js {
		log.Debugf("Waiting for Unit(%s) state(%s) to be %s", name, job.JobState(u.CurrentState), js)
		return
	}

	ret = true
	msg := fmt.Sprintf("Unit %s %s", name, u.CurrentState)

	if u.MachineID != "" {
		ms := cachedMachineState(u.MachineID)
		if ms != nil {
			msg = fmt.Sprintf("%s on %s", msg, machineFullLegend(*ms, false))
		}
	}

	fmt.Fprintln(out, msg)
	return
}

func globMatches(pattern, target string) bool {
	matched, err := path.Match(pattern, target)
	if err != nil {
		log.Debugf("Received error while matching pattern '%s': %v", pattern, err)
	}
	return matched
}

// getUnitFileFromTemplate attempts to get a Unit from a template unit that
// is either in the registry or on the file system
// It takes two arguments, the template information and the unit file name
// It returns the Unit or nil; and any error encountered
func getUnitFileFromTemplate(cCmd *cobra.Command, uni *unit.UnitNameInfo, fileName string) (*unit.UnitFile, error) {
	var uf *unit.UnitFile

	tmpl, err := cAPI.Unit(uni.Template)
	if err != nil {
		return nil, fmt.Errorf("error retrieving template Unit(%s) from Registry: %v", uni.Template, err)
	}

	if tmpl != nil {
		isLocalUnitDifferent(cCmd, fileName, tmpl, false)
		uf = schema.MapSchemaUnitOptionsToUnitFile(tmpl.Options)
		log.Debugf("Template Unit(%s) found in registry", uni.Template)
	} else {
		// Finally, if we could not find a template unit in the Registry,
		// check the local disk for one instead
		filePath := path.Join(path.Dir(fileName), uni.Template)
		if _, err := os.Stat(filePath); os.IsNotExist(err) {
			return nil, fmt.Errorf("unable to find template Unit(%s) in Registry or on filesystem", uni.Template)
		}

		uf, err = getUnitFromFile(filePath)
		if err != nil {
			return nil, fmt.Errorf("unable to load template Unit(%s) from file: %v", uni.Template, err)
		}
	}

	return uf, nil
}

func acquireLeadership(lManager lease.Manager, machID string, ver int, ttl time.Duration) lease.Lease {
	existing, err := lManager.GetLease(engineLeaseName)
	if err != nil {
		log.Errorf("Unable to determine current lease: %v", err)
		return nil
	}

	var l lease.Lease
	if existing == nil {
		l, err = lManager.AcquireLease(engineLeaseName, machID, ver, ttl)
		if err != nil {
			log.Errorf("Engine leadership acquisition failed: %v", err)
			return nil
		} else if l == nil {
			log.Debugf("Unable to acquire engine leadership")
			return nil
		}
		log.Infof("Engine leadership acquired")
		metrics.ReportEngineLeader()
		return l
	}

	if existing.Version() >= ver {
		log.Debugf("Lease already held by Machine(%s) operating at acceptable version %d", existing.MachineID(), existing.Version())
		return existing
	}

	rem := existing.TimeRemaining()
	l, err = lManager.StealLease(engineLeaseName, machID, ver, ttl+rem, existing.Index())
	if err != nil {
		log.Errorf("Engine leadership steal failed: %v", err)
		return nil
	} else if l == nil {
		log.Debugf("Unable to steal engine leadership")
		return nil
	}

	log.Infof("Stole engine leadership from Machine(%s)", existing.MachineID())
	metrics.ReportEngineLeader()

	if rem > 0 {
		log.Infof("Waiting %v for previous lease to expire before continuing reconciliation", rem)
		<-time.After(rem)
	}

	return l
}

func NewCoreOSMachine(static MachineState, um unit.UnitManager) *CoreOSMachine {
	log.Debugf("Created CoreOSMachine with static state %s", static)
	m := &CoreOSMachine{
		staticState: static,
		um:          um,
	}
	return m
}

// desiredAgentState builds an *AgentState object that represents what the
// provided Agent should currently be doing.
func desiredAgentState(a *Agent, reg registry.Registry) (*AgentState, error) {
	units, err := reg.Units()
	if err != nil {
		log.Errorf("Failed fetching Units from Registry: %v", err)
		return nil, err
	}

	sUnits, err := reg.Schedule()
	if err != nil {
		log.Errorf("Failed fetching schedule from Registry: %v", err)
		return nil, err
	}

	// fetch full machine state from registry instead of
	// using the local version to allow for dynamic metadata
	ms, err := reg.MachineState(a.Machine.State().ID)
	if err != nil {
		log.Errorf("Failed fetching machine state from Registry: %v", err)
		return nil, err
	}
	as := AgentState{
		MState: &ms,
		Units:  make(map[string]*job.Unit),
	}

	sUnitMap := make(map[string]*job.ScheduledUnit)
	for _, sUnit := range sUnits {
		sUnit := sUnit
		sUnitMap[sUnit.Name] = &sUnit
	}

	for _, u := range units {
		u := u
		md := u.RequiredTargetMetadata()

		if u.IsGlobal() {
			if !machine.HasMetadata(&ms, md) {
				log.Debugf("Agent unable to run global unit %s: missing required metadata", u.Name)
				continue
			}
		}

		if !u.IsGlobal() {
			sUnit, ok := sUnitMap[u.Name]
			if !ok || sUnit.TargetMachineID == "" || sUnit.TargetMachineID != ms.ID {
				continue
			}
		}

		if cExists, _ := as.HasConflict(u.Name, u.Conflicts()); cExists {
			continue
		}

		as.Units[u.Name] = &u
	}

	return &as, nil
}

func renewLeadership(l lease.Lease, ttl time.Duration) lease.Lease {
	err := l.Renew(ttl)
	if err != nil {
		log.Errorf("Engine leadership lost, renewal failed: %v", err)
		return nil
	}

	log.Debugf("Engine leadership renewed")
	return l
}