// WriteLock writes the lock as YAML.
//
// Params:
//	- lockfile: A *cfg.Lockfile to render.
// 	- out (io.Writer): An output stream to write to. Default is os.Stdout.
func WriteLock(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	skip := p.Get("skip", false).(bool)
	if skip {
		return false, nil
	}

	lockfile := p.Get("lockfile", nil).(*cfg.Lockfile)

	Info("Writing glide.lock file")

	data, err := lockfile.Marshal()
	if err != nil {
		return nil, err
	}

	var out io.Writer
	file, err := os.Create("glide.lock")
	if err != nil {
		return false, err
	}
	defer file.Close()
	out = io.Writer(file)
	out.Write(data)

	return true, nil
}

// ExecCmd executes a system command  inside vendor
func ExecCmd(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	args := p.Get("args", nil).(cli.Args)

	if len(args) == 0 {
		return nil, fmt.Errorf("No command to execute")
	}

	gopath, err := VendorPath(c)
	if err != nil {
		return false, err
	}

	err = os.Setenv("GOPATH", gopath)
	if err != nil {
		return false, err
	}

	path := os.Getenv("PATH")
	err = os.Setenv("PATH", gopath+"/bin:"+path)
	if err != nil {
		return false, err
	}

	cmd := exec.Command(args[0], args[1:]...)
	cmd.Stdout = os.Stdout
	cmd.Stderr = os.Stderr
	err = cmd.Run()

	return true, nil
}

// MergeToYaml converts a Config object and a yaml.File to a single yaml.File.
//
// Params:
//	- conf (*Config): The configuration to merge.
//	- overwriteImports (bool, default true): If this is true, old config will
//		overwritten. If false, we attempt to merge the old and new config, with
//		preference to the old.
//
// Returns:
//	- The root yaml.Node of the modified config.
//
// Uses:
//	- cxt.Get("yaml.File") as the source for the YAML file.
func MergeToYaml(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	root := c.Get("yaml.File", nil).(*yaml.File).Root
	cfg := p.Get("conf", nil).(*Config)
	overwrite := p.Get("overwriteImports", true).(bool)

	rootMap, ok := root.(yaml.Map)
	if !ok {
		return nil, fmt.Errorf("Expected root node to be a map.")
	}

	if len(cfg.Name) > 0 {
		rootMap["package"] = yaml.Scalar(cfg.Name)
	}
	if cfg.InCommand != "" {
		rootMap["incmd"] = yaml.Scalar(cfg.InCommand)
	}

	if overwrite {
		// Imports
		imports := make([]yaml.Node, len(cfg.Imports))
		for i, imp := range cfg.Imports {
			imports[i] = imp.ToYaml()
		}
		rootMap["import"] = yaml.List(imports)
	} else {
		var err error
		rootMap, err = mergeImports(rootMap, cfg)
		if err != nil {
			Warn("Problem merging imports: %s\n", err)
		}
	}

	return root, nil
}

/**
 * Perform authentication.
 *
 * Params:
 * 	- realm (string): The name of the realm. (Default: "web")
 * 	- datasource (string): The name of the datasource that should be used to authenticate.
 * 	  This datasource must be an `auth.UserDatasource`. (Default: "auth.UserDatasource")
 *
 * Context:
 * 	- http.Request (*http.Request): The HTTP request. This is usually placed into the
 * 	context for you.
 * 	- http.ResponseWriter (http.ResponseWriter): The response. This is usually placed
 * 	 into the context for you.
 *
 * Datasource:
 * 	- An auth.UserDatasource. By default, this will look for a datasource named
 * 	  "auth.UserDatasource". This can be overridden by the `datasource` param.
 *
 * Returns:
 * 	- True if the user authenticated. If not, this will send a 401 and then stop
 * 	  the current chain.
 */
func Basic(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	realm := p.Get("realm", "web").(string)
	dsName := p.Get("datasource", "auth.UserDatasource").(string)

	req := c.Get("http.Request", nil).(*http.Request)
	res := c.Get("http.ResponseWriter", nil).(http.ResponseWriter)

	ds := c.Datasource(dsName).(UserDatasource)

	authz := strings.TrimSpace(req.Header.Get("Authorization"))
	if len(authz) == 0 || !strings.Contains(authz, "Basic ") {
		return sendUnauthorized(realm, res)
	}

	user, pass, err := parseBasicString(authz)
	if err != nil {
		c.Logf("info", "Basic authentication parsing failed: %s", err)
		return sendUnauthorized(realm, res)
	}

	ok, err := ds.AuthUser(user, pass)
	if !ok {
		if err != nil {
			c.Logf("info", "Basic authentication caused an error: %s", err)
		}
		return sendUnauthorized(realm, res)
	}

	return ok, err
}

// ListDeps lists all of the dependencies of the current project.
//
// Params:
//
// Returns:
//
func ListDeps(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	buildContext, err := GetBuildContext()
	if err != nil {
		return nil, err
	}
	basedir := p.Get("dir", ".").(string)
	myName := guessPackageName(buildContext, basedir)

	basedir, err = filepath.Abs(basedir)
	if err != nil {
		return nil, err
	}

	direct := map[string]*pinfo{}
	d := walkDeps(buildContext, basedir, myName)
	for _, i := range d {
		listDeps(buildContext, direct, i, basedir)
	}

	sortable := make([]string, len(direct))
	i := 0
	for k := range direct {
		sortable[i] = k
		i++
	}

	sort.Strings(sortable)

	for _, k := range sortable {
		t := direct[k].PType
		fmt.Printf("%s (Location: %s)\n", k, ptypeString(t))
	}

	return nil, nil
}

// ParallelBuild runs multiple docker builds at the same time.
//
// Params:
//	-images ([]BuildImg): Images to build
// 	-alwaysFetch (bool): Default false. If set to true, this will always fetch
// 		the Docker image even if it already exists in the registry.
//
// Returns:
//
// 	- Waiter: A *sync.WaitGroup that is waiting for the docker downloads to finish.
//
// Context:
//
// This puts 'ParallelBuild.failN" (int) into the context to indicate how many failures
// occurred during fetches.
func ParallelBuild(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	images := p.Get("images", []BuildImg{}).([]BuildImg)

	var wg sync.WaitGroup
	var m sync.Mutex
	var fails int

	for _, img := range images {
		img := img

		// HACK: ensure "docker build" is serialized by allowing only one entry in
		// the WaitGroup. This works around the "simultaneous docker pull" bug.
		wg.Wait()
		wg.Add(1)
		safely.GoDo(c, func() {
			log.Infof(c, "Starting build for %s (tag: %s)", img.Path, img.Tag)
			if _, err := buildImg(c, img.Path, img.Tag); err != nil {
				log.Errf(c, "Failed to build docker image: %s", err)
				m.Lock()
				fails++
				m.Unlock()
			}
			wg.Done()
		})

	}

	// Number of failures.
	c.Put("ParallelBuild.failN", fails)

	return &wg, nil
}

// Watch watches a given path, and executes a git check-repos for each event.
//
// It starts the watcher and then returns. The watcher runs on its own
// goroutine. To stop the watching, send the returned channel a bool.
//
// Params:
// - client (Watcher): An Etcd client.
// - path (string): The path to watch
//
// Returns:
// 	- chan bool: Send this a message to stop the watcher.
func Watch(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	// etcdctl -C $ETCD watch --recursive /deis/services
	path := p.Get("path", "/deis/services").(string)
	cli, ok := p.Has("client")
	if !ok {
		return nil, errors.New("No etcd client found.")
	}
	client := cli.(Watcher)

	// Stupid hack because etcd watch seems to be broken, constantly complaining
	// that the JSON it received is malformed.
	safely.GoDo(c, func() {
		for {
			response, err := client.Watch(path, 0, true, nil, nil)
			if err != nil {
				log.Errf(c, "Etcd Watch failed: %s", err)
				time.Sleep(50 * time.Millisecond)
				continue
			}

			if response.Node == nil {
				log.Infof(c, "Unexpected Etcd message: %v", response)
			}
			git := exec.Command("/home/git/check-repos")
			if out, err := git.CombinedOutput(); err != nil {
				log.Errf(c, "Failed git check-repos: %s", err)
				log.Infof(c, "Output: %s", out)
			}
		}

	})

	return nil, nil

}

// ListDeps lists all of the dependencies of the current project.
//
// Params:
//
// Returns:
//
func ListDeps(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	basedir := p.Get("dir", ".").(string)
	myName := guessPackageName(basedir)

	var err error
	basedir, err = filepath.Abs(basedir)
	if err != nil {
		return nil, err
	}

	direct := map[string]bool{}
	d := walkDeps(basedir, myName)
	for _, i := range d {
		listDeps(direct, i, basedir)
	}

	sortable := make([]string, len(direct))
	i := 0
	for k, _ := range direct {
		sortable[i] = k
		i++
	}

	sort.Strings(sortable)

	for _, k := range sortable {
		dec := "no"
		if direct[k] {
			dec = "yes"
		}
		fmt.Printf("%s (Present: %s)\n", k, dec)
	}

	return nil, nil
}

// Subscribe allows an request to subscribe to topic updates.
//
// Params:
// 	- topic (string): The topic to subscribe to.
// 	-
//
// Returns:
//
func Subscribe(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	medium, err := getMedium(c)
	if err != nil {
		return nil, &cookoo.FatalError{"No medium."}
	}
	topic := p.Get("topic", "").(string)
	if len(topic) == 0 {
		return nil, errors.New("No topic is set.")
	}

	rw := c.Get("http.ResponseWriter", nil).(ResponseWriterFlusher)
	clientGone := rw.(http.CloseNotifier).CloseNotify()

	sub := NewSubscription(rw)
	t := fetchOrCreateTopic(medium, topic, true, DefaultMaxHistory)
	t.Subscribe(sub)

	defer func() {
		t.Unsubscribe(sub)
		sub.Close()
	}()

	sub.Listen(clientGone)

	return nil, nil
}

// Watch watches a given path, and executes a git check-repos for each event.
//
// It starts the watcher and then returns. The watcher runs on its own
// goroutine. To stop the watching, send the returned channel a bool.
//
// Params:
// - client (Watcher): An Etcd client.
// - path (string): The path to watch
func Watch(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {

	// etcdctl -C $ETCD watch --recursive /deis/services
	path := p.Get("path", "/deis/services").(string)
	cli := p.Get("client", nil).(client.Client)
	k := client.NewKeysAPI(cli)

	watcher := k.Watcher(path, &client.WatcherOptions{Recursive: true})

	safely.GoDo(c, func() {
		for {
			// TODO: We should probably add cancellation support.
			response, err := watcher.Next(dctx())
			if err != nil {
				log.Errf(c, "Etcd Watch failed: %s", err)
			}
			if response.Node == nil {
				log.Infof(c, "Unexpected Etcd message: %v", response)
			}
			git := exec.Command("/home/git/check-repos")
			if out, err := git.CombinedOutput(); err != nil {
				log.Errf(c, "Failed git check-repos: %s", err)
				log.Infof(c, "Output: %s", out)
			}
		}
	})
	return nil, nil
}

// ParseHostKeys parses the host key files.
//
// By default it looks in /etc/ssh for host keys of the patterh ssh_host_{{TYPE}}_key.
//
// Params:
// 	- keytypes ([]string): Key types to parse. Defaults to []string{rsa, dsa, ecdsa}
// 	- enableV1 (bool): Allow V1 keys. By default this is disabled.
// 	- path (string): Override the lookup pattern. If %s, it will be replaced with the keytype.
//
// Returns:
// 	[]ssh.Signer
func ParseHostKeys(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	log.Debugf(c, "Parsing ssh host keys")
	hostKeyTypes := p.Get("keytypes", []string{"rsa", "dsa", "ecdsa"}).([]string)
	pathTpl := p.Get("path", "/etc/ssh/ssh_host_%s_key").(string)
	hostKeys := make([]ssh.Signer, 0, len(hostKeyTypes))
	for _, t := range hostKeyTypes {
		path := fmt.Sprintf(pathTpl, t)

		if key, err := ioutil.ReadFile(path); err == nil {
			if hk, err := ssh.ParsePrivateKey(key); err == nil {
				log.Infof(c, "Parsed host key %s.", path)
				hostKeys = append(hostKeys, hk)
			} else {
				log.Errf(c, "Failed to parse host key %s (skipping): %s", path, err)
			}
		}
	}
	if c.Get("enableV1", false).(bool) {
		path := "/etc/ssh/ssh_host_key"
		if key, err := ioutil.ReadFile(path); err != nil {
			log.Errf(c, "Failed to read ssh_host_key")
		} else if hk, err := ssh.ParsePrivateKey(key); err == nil {
			log.Infof(c, "Parsed host key %s.", path)
			hostKeys = append(hostKeys, hk)
		} else {
			log.Errf(c, "Failed to parse host key %s: %s", path, err)
		}
	}
	return hostKeys, nil
}

// Push pushes an image to the registry.
//
// This finds the appropriate registry by looking it up in etcd.
//
// Params:
// - client (etcd.Getter): Client to do etcd lookups.
// - tag (string): Tag to push.
//
// Returns:
//
func Push(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	// docker tag deis/slugrunner:lastest HOST:PORT/deis/slugrunner:latest
	// docker push HOST:PORT/deis/slugrunner:latest
	client := p.Get("client", nil).(etcd.Getter)

	host, err := client.Get("/deis/registry/host", false, false)
	if err != nil || host.Node == nil {
		return nil, err
	}
	port, err := client.Get("/deis/registry/port", false, false)
	if err != nil || host.Node == nil {
		return nil, err
	}

	registry := host.Node.Value + ":" + port.Node.Value
	tag := p.Get("tag", "").(string)

	log.Infof(c, "Pushing %s to %s. This may take some time.", tag, registry)
	rem := path.Join(registry, tag)

	out, err := exec.Command("docker", "tag", "-f", tag, rem).CombinedOutput()
	if err != nil {
		log.Warnf(c, "Failed to tag %s on host %s: %s (%s)", tag, rem, err, out)
	}
	out, err = exec.Command("docker", "-D", "push", rem).CombinedOutput()

	if err != nil {
		log.Warnf(c, "Failed to push %s to host %s: %s (%s)", tag, rem, err, out)
		return nil, err
	}
	log.Infof(c, "Finished pushing %s to %s.", tag, registry)
	return nil, nil
}

// Template is a template-based text formatter.
//
// This uses the core `text/template` to process a given string template.
//
// Params
// 	- template (string): A template string.
// 	- template.Context (bool): If true, the context will be placed into the
// 		template renderer as 'Cxt', and can be used as `{{.Cxt.Foo}}`. False
// 		by default.
// 	- ... (interface{}): Values passed into the template.
//
// Conventionally, template variables should start with an initial capital.
//
// Returns a formatted string.
func Template(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	format := cookoo.GetString("template", "", p)
	withCxt := cookoo.GetBool("template.Context", false, p)

	name := fmt.Sprintf("%x", md5.Sum([]byte(format)))

	//c.Logf("debug", "Template %s is '%s'\n", name, format)

	tpl, err := template.New(name).Parse(format)
	if err != nil {
		return "", err
	}

	data := p.AsMap()
	if withCxt {
		//c.Logf("debug", "Adding context.")
		data["Cxt"] = c.AsMap()
	}

	var out bytes.Buffer
	if err := tpl.Execute(&out, data); err != nil {
		return "", err
	}

	return out.String(), nil
}

// ParallelBuild runs multiple docker builds at the same time.
//
// Params:
//	-images ([]BuildImg): Images to build
// 	-alwaysFetch (bool): Default false. If set to true, this will always fetch
// 		the Docker image even if it already exists in the registry.
//
// Returns:
//
// 	- Waiter: A *sync.WaitGroup that is waiting for the docker downloads to finish.
//
// Context:
//
// This puts 'ParallelBuild.failN" (int) into the context to indicate how many failures
// occurred during fetches.
func ParallelBuild(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	images := p.Get("images", []BuildImg{}).([]BuildImg)

	var wg sync.WaitGroup
	var m sync.Mutex
	var fails int

	for _, img := range images {
		img := img
		wg.Add(1)
		safely.GoDo(c, func() {
			log.Infof(c, "Starting build for %s (tag: %s)", img.Path, img.Tag)
			if _, err := buildImg(c, img.Path, img.Tag); err != nil {
				log.Errf(c, "Failed to build docker image: %s", err)
				m.Lock()
				fails++
				m.Unlock()
			}
			wg.Done()
		})

	}

	// Number of failures.
	c.Put("ParallelBuild.failN", fails)

	return &wg, nil
}

// RemoveMemberByName removes a member whose name matches the given.
//
// Params:
// 	- client(client.Client): An etcd client
//	- name (string): The name to remove
// Returns:
//	true if the member was found, false otherwise.
func RemoveMemberByName(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	cli := p.Get("client", nil).(client.Client)
	name := p.Get("name", "____").(string)
	mem := client.NewMembersAPI(cli)

	members, err := mem.List(dctx())
	if err != nil {
		log.Errf(c, "Could not get a list of members: %s", err)
		return false, err
	}

	remIDs := []string{}
	for _, member := range members {
		if member.Name == name {
			log.Infof(c, "Removing member %s (ID: %s)", name, member.ID)
			// If this is synchronizable, we should do it in parallel.
			if err := mem.Remove(dctx(), member.ID); err != nil {
				log.Errf(c, "Failed to remove member: %s", err)
				return len(remIDs) > 0, err
			}
			remIDs = append(remIDs, member.ID)
		}
	}

	return len(remIDs) > 0, nil
}

// FindSSHUser finds an SSH user by public key.
//
// Some parts of the system require that we know not only the SSH key, but also
// the name of the user. That information is stored in etcd.
//
// Params:
// 	- client (EtcdGetter)
// 	- fingerprint (string): The fingerprint of the SSH key.
//
// Returns:
// - username (string)
func FindSSHUser(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	client := p.Get("client", nil).(Getter)
	fingerprint := p.Get("fingerprint", nil).(string)

	res, err := client.Get("/deis/builder/users", false, true)
	if err != nil {
		log.Warnf(c, "Error querying etcd: %s", err)
		return "", err
	} else if res.Node == nil || !res.Node.Dir {
		log.Warnf(c, "No users found in etcd.")
		return "", errors.New("Users not found")
	}
	for _, user := range res.Node.Nodes {
		log.Infof(c, "Checking user %s", user.Key)
		for _, keyprint := range user.Nodes {
			if strings.HasSuffix(keyprint.Key, fingerprint) {
				parts := strings.Split(user.Key, "/")
				username := parts[len(parts)-1]
				log.Infof(c, "Found user %s for fingerprint %s", username, fingerprint)
				return username, nil
			}
		}
	}

	return "", fmt.Errorf("User not found for fingerprint %s", fingerprint)
}

// VendoredCleanUp is a command that cleans up vendored codebases after an update.
// If enabled (via update) it removed the VCS info from updated vendored
// packages. This should be a suffix to UpdateImports and  VendoredSetup should
// be a prefix to UpdateImports.
func VendoredCleanUp(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	update := p.Get("update", true).(bool)
	if update != true {
		return false, nil
	}
	cfg := p.Get("conf", nil).(*Config)

	vend, err := VendorPath(c)
	if err != nil {
		return false, err
	}

	for _, dep := range cfg.Imports {
		if dep.UpdateAsVendored == true {
			Info("Cleaning up vendored package %s\n", dep.Name)

			// Remove the VCS directory
			cwd := path.Join(vend, dep.Name)
			repo, err := dep.GetRepo(cwd)
			if err != nil {
				Error("Error cleaning up %s:%s", dep.Name, err)
				continue
			}
			t := repo.Vcs()
			err = os.RemoveAll(cwd + string(os.PathSeparator) + "." + string(t))
			if err != nil {
				Error("Error cleaning up VCS dir for %s:%s", dep.Name, err)
			}
		}

	}

	return true, nil
}

// UpdateReferences updates the revision numbers on all of the imports.
//
// If a `packages` list is supplied, only the given base packages will
// be updated.
//
// Params:
// 	- conf (*yaml.Config): Configuration
// 	- packages ([]string): A list of packages to update. Default is all packages.
func UpdateReferences(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	cfg := p.Get("conf", &yaml.Config{}).(*yaml.Config)
	plist := p.Get("packages", []string{}).([]string)

	pkgs := list2map(plist)
	restrict := len(pkgs) > 0

	cwd, err := VendorPath(c)
	if err != nil {
		return false, err
	}

	if len(cfg.Imports) == 0 {
		return cfg, nil
	}

	for _, imp := range cfg.Imports {
		if restrict && !pkgs[imp.Name] {
			Debug("===> Skipping %q", imp.Name)
			continue
		}
		commit, err := VcsLastCommit(imp, cwd)
		if err != nil {
			Warn("Could not get commit on %s: %s", imp.Name, err)
		}
		imp.Reference = commit
	}

	return cfg, nil
}

// RunOnce runs the equivalent of `confd --onetime`.
//
// This may run the process repeatedly until either we time out (~20 minutes) or
// the templates are successfully built.
//
// Importantly, this blocks until the run is complete.
//
// Params:
// - node (string): The etcd node to use. (Only etcd is currently supported)
//
// Returns:
// - The []bytes from stdout and stderr when running the program.
//
func RunOnce(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	node := p.Get("node", defaultEtcd).(string)

	dargs := []string{"-onetime", "-node", node, "-log-level", "error"}

	log.Info(c, "Building confd templates. This may take a moment.")

	limit := 1200
	timeout := time.Second * 3
	var lasterr error
	start := time.Now()
	for i := 0; i < limit; i++ {
		out, err := exec.Command("confd", dargs...).CombinedOutput()
		if err == nil {
			log.Infof(c, "Templates generated for %s on run %d", node, i)
			return out, nil
		}
		log.Debugf(c, "Recoverable error: %s", err)
		log.Debugf(c, "Output: %q", out)
		lasterr = err

		time.Sleep(timeout)
		log.Infof(c, "Re-trying template build. (Elapsed time: %d)", time.Now().Sub(start)/time.Second)
	}

	return nil, fmt.Errorf("Could not build confd templates before timeout. Last error: %s", lasterr)
}

// ReplayHistory sends back the history to a subscriber.
//
// This should be called before the client goes into active listening.
//
// Params:
// - topic (string): The topic to fetch.
//
// Returns:
// 	- int: The number of history messages sent to the client.
func ReplayHistory(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	req := c.Get("http.Request", nil).(*http.Request)
	res := c.Get("http.ResponseWriter", nil).(ResponseWriterFlusher)
	medium, _ := getMedium(c)
	name := p.Get("topic", "").(string)

	// This does not manage topics. If there is no topic set, we silently fail.
	if len(name) == 0 {
		c.Log("info", "No topic name given to ReplayHistory.")
		return 0, nil
	}
	top, ok := medium.Topic(name)
	if !ok {
		c.Logf("info", "No topic named %s exists yet. No history replayed.", name)
		return 0, nil
	}

	topic, ok := top.(HistoriedTopic)
	if !ok {
		c.Logf("info", "No history for topic %s.", name)
		res.Header().Add(XHistoryEnabled, "False")
		return 0, nil
	}
	res.Header().Add(XHistoryEnabled, "True")

	since := req.Header.Get(XHistorySince)
	max := req.Header.Get(XHistoryLength)

	// maxLen can be used either on its own or paired with X-History-Since.
	maxLen := 0
	if len(max) > 0 {
		m, err := parseHistLen(max)
		if err != nil {
			c.Logf("info", "failed to parse X-History-Length %s", max)
		} else {
			maxLen = m
		}
	}
	if len(since) > 0 {
		ts, err := parseSince(since)
		if err != nil {
			c.Logf("warn", "Failed to parse X-History-Since field %s: %s", since, err)
			return 0, nil
		}
		toSend := topic.Since(ts)

		// If maxLen is also set, we trim the list by sending the newest.
		ls := len(toSend)
		if maxLen > 0 && ls > maxLen {
			offset := ls - maxLen - 1
			toSend = toSend[offset:]
		}
		return sendHistory(c, res, toSend)
	} else if maxLen > 0 {
		toSend := topic.Last(maxLen)
		return sendHistory(c, res, toSend)
	}

	return 0, nil
}