// RemoveFromList retrieves the specified key from the data store and removes
// the specified value from its list. If the key does not fall within the
// receiving server's range, it should reply with status WrongServer. If
// the specified value is not already contained in the list, it should reply
// with status ItemNotFound.
func (ss *storageServer) RemoveFromList(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	//	fmt.Println("RemoveFromList function called:", args);
	if FindPartitionNode(ss.ring, args.Key).NodeID != ss.nodeID {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	ss.sMutex.Lock()
	data, ok := ss.storage[args.Key]
	ss.sMutex.Unlock()
	if !ok {
		reply.Status = storagerpc.KeyNotFound
	} else {
		data.dMutex.Lock()

		list := data.data.([]string)
		index := ss.ListContains(list, args.Value)
		if index == -1 {
			reply.Status = storagerpc.ItemNotFound
		} else {
			reply.Status = storagerpc.OK
			ss.RevokeCallBacksForKey(args.Key, data)
			list = append(list[:index], list[index+1:]...)
			data.data = list
		}

		data.dMutex.Unlock()
	}

	return nil
}

func (ss *storageServer) Put(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {

	key := args.Key

	if !correctServer(ss, key) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	leases := getLeases(ss, key)
	if leases != nil {
		ss.editLock.Lock()
		revokeLease(ss, key)
		ss.editLock.Unlock()
	}

	ss.valMapLock.Lock()
	defer ss.valMapLock.Unlock()

	value := args.Value
	ss.valMap[key] = value
	reply.Status = storagerpc.OK
	return nil

}

/// TODO: add the lease function
func (ss *storageServer) Put(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {

	//	fmt.Println("Put function is called.", args);
	if FindPartitionNode(ss.ring, args.Key).NodeID != ss.nodeID {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	ss.sMutex.Lock()
	data, ok := ss.storage[args.Key]
	if ok {
		ss.sMutex.Unlock()
		//		fmt.Println("Put function called: Update a value")
		reply.Status = storagerpc.OK

		// need to deal with the revoke
		data.dMutex.Lock()
		ss.RevokeCallBacksForKey(args.Key, data)

		// apply changes
		// note that the data might be already deleted, so not in the storage map..
		data.data = args.Value
		data.dMutex.Unlock()
	} else {
		//		fmt.Println("Put function called: Insert a value")
		reply.Status = storagerpc.OK
		data = &dataUnit{data: args.Value, Set: nil}
		ss.storage[args.Key] = data

		ss.sMutex.Unlock()
	}
	return nil
}

func (ss *storageServer) Put(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	if ss.isReady == false {
		reply.Status = storagerpc.NotReady
	} else {
		if ss.getServerID(args.Key) != ss.nodeID {
			reply.Status = storagerpc.WrongServer
			return nil
		}
		ss.serverMutex.Lock()
		if ss.userValueMutex[args.Key] == nil {
			ss.userValueMutex[args.Key] = &sync.Mutex{}
		}
		ss.serverMutex.Unlock()
		ss.userValueMutex[args.Key].Lock()
		defer ss.userValueMutex[args.Key].Unlock()
		_, exist := ss.userValue[args.Key]
		newMap := make(map[string]*lease)
		if exist == true {
			for k, v := range ss.userValueLease[args.Key] {
				newMap[k] = v
			}
			// revoke all leases if already exist in server before
			err := ss.revokeLease(args.Key, newMap)
			if err != nil {
				return err
			}
		}
		// make new lease map
		ss.userValueLease[args.Key] = make(map[string]*lease)
		ss.userValue[args.Key] = args.Value
	}
	reply.Status = storagerpc.OK
	return nil
}

func (ss *storageServer) RemoveFromList(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	status, ok := ss.assertKeyAndServer(args.Key)
	if !ok {
		reply.Status = status
		return nil
	}

	ss.rwl.Lock()
	defer ss.rwl.Unlock()

	ss.waitForRevoking(args.Key)

	if _, exisit := ss.store[args.Key]; !exisit {
		reply.Status = storagerpc.KeyNotFound
		return nil
	}

	// test item exisitence
	l := ss.store[args.Key].(*list.List)
	for e := l.Front(); e != nil; e = e.Next() {
		if e.Value.(string) == args.Value {
			ss.revokeLease(args.Key)
			l.Remove(e)
			reply.Status = storagerpc.OK
			return nil
		}
	}

	// item not found
	reply.Status = storagerpc.ItemNotFound
	return nil
}

func (ss *storageServer) Put(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	//fmt.Println("Put value: ", args.Value, " to key ", args.Key)
	if !ss.IsKeyInRange(args.Key) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	ss.lockForTableLocks.Lock()
	/* Lock table for writing */
	lock, exist := ss.tableLocks[args.Key]
	if !exist {
		lock = new(sync.Mutex)
		ss.tableLocks[args.Key] = lock
	}
	ss.lockForTableLocks.Unlock()

	lock.Lock()
	defer lock.Unlock()

	ss.RevokeLease(args.Key)

	ss.stringTable[args.Key] = args.Value
	reply.Status = storagerpc.OK

	return nil
}

func (ss *storageServer) AppendToList(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	if args == nil {
		return errors.New("ss: Can't append nil K/V pair")
	}
	if reply == nil {
		return errors.New("ss: Can't reply with nil in Append")
	}
	if !(ss.CheckKeyInRange(args.Key)) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	ss.sMutex.Lock()
	keyLock, exist := ss.keyLockMap[args.Key]
	if !exist {
		// Create new lock for the key
		keyLock = &sync.Mutex{}
		ss.keyLockMap[args.Key] = keyLock
	}
	ss.sMutex.Unlock()

	keyLock.Lock()
	defer keyLock.Unlock()

	hpTimeMap, leaseExists := ss.leaseMap[args.Key]
	if leaseExists {
		// Revoke all issued leases.
		successChan := make(chan int, 1)
		finishChan := make(chan int, 1)
		expected := len(ss.leaseMap[args.Key])
		go ss.CheckRevokeStatus(args.Key, successChan, finishChan, expected)
		for hp, _ := range hpTimeMap {
			go ss.RevokeLeaseAt(hp, args.Key, successChan)
		}

		<-finishChan

		delete(ss.leaseMap, args.Key)
	}

	lst, ok := ss.lMap[args.Key]
	if ok {
		for _, v := range lst.value {
			if v == args.Value {
				reply.Status = storagerpc.ItemExists
				return nil
			}
		}
		lst.value = append(lst.value, args.Value)
	} else {
		newValue := lValue{
			value: make([]string, 1),
		}
		newValue.value[0] = args.Value
		ss.lMap[args.Key] = &newValue
	}
	reply.Status = storagerpc.OK
	return nil
}

func (ss *storageServer) RemoveFromList(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	if args == nil {
		return errors.New("ss: Can't Reomove nil K/V pair")
	}
	if reply == nil {
		return errors.New("ss: Can't reply with nil in Remove")
	}
	if !(ss.CheckKeyInRange(args.Key)) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	ss.sMutex.Lock()
	keyLock, exist := ss.keyLockMap[args.Key]
	if !exist {
		// Create new lock for the key
		keyLock = &sync.Mutex{}
		ss.keyLockMap[args.Key] = keyLock
	}
	ss.sMutex.Unlock()

	keyLock.Lock()
	defer keyLock.Unlock()

	hpTimeMap, leaseExists := ss.leaseMap[args.Key]
	if leaseExists {
		// Revoke all issued leases.
		successChan := make(chan int, 1)
		finishChan := make(chan int, 1)
		expected := len(ss.leaseMap[args.Key])
		go ss.CheckRevokeStatus(args.Key, successChan, finishChan, expected)
		for hp, _ := range hpTimeMap {
			go ss.RevokeLeaseAt(hp, args.Key, successChan)
		}

		<-finishChan

		delete(ss.leaseMap, args.Key)
	}

	lst, ok := ss.lMap[args.Key]
	if ok {
		for i, v := range lst.value {
			if v == args.Value {
				lst.value = append(lst.value[:i], lst.value[i+1:]...)
				reply.Status = storagerpc.OK
				return nil
			}
		}
	}
	reply.Status = storagerpc.ItemNotFound
	return nil
}

func (ss *storageServer) AppendToList(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {

	key := args.Key

	if rightStorageServer(ss, key) == false {
		reply.Status = storagerpc.WrongServer
		ss.keyLocks[key] <- 1

		return nil
	}

	if _, found := ss.keyLocks[key]; found == false {
		ss.keyLocks[key] = make(chan int, 1)
	} else {
		<-ss.keyLocks[key]
	}

	if lst, found := ss.topMap[key]; found {
		if l, ok := lst.([]string); ok {

			for i := 0; i < len(l); i++ {
				if l[i] == args.Value {
					// value was already in list
					reply.Status = storagerpc.ItemExists
					ss.keyLocks[key] <- 1

					return nil
				}
			}
			// value was not in list, append to the end

			reply.Status = storagerpc.OK

			ss.topMap[key] = append(l, args.Value)
		} else {
			// list was corrputed, shouldn't happen
			ss.keyLocks[key] <- 1

			return errors.New("List to remove from is wrong type")
		}
	} else {
		// This key hasn't had a list made yet, make new list and insert value
		l := make([]string, 1)
		l[0] = args.Value

		ss.topMap[key] = l
		reply.Status = storagerpc.OK
	}
	ss.keyLocks[key] <- 1

	return nil
}

func (ss *storageServer) Put(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	if !ss.isInServerRange(args.Key) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	//get the lock for current key
	ss.mutex.Lock()
	lock, exist := ss.lockMap[args.Key]
	if !exist {
		lock = new(sync.Mutex)
		ss.lockMap[args.Key] = lock
	}
	ss.mutex.Unlock()

	//Lock current key we are going to work on
	lock.Lock()

	//If has granted lease for current key, then revoke all the lease
	_, exist = ss.leaseMap[args.Key]
	if exist {
		//revoke all the granted Lease
		finishRevokeChan := make(chan struct{})
		singleRevokeChan := make(chan struct{})
		leaseHolderList := ss.leaseMap[args.Key]
		//revokeNum := leaseHolderList.Len()

		go ss.revokeHandler(args.Key, singleRevokeChan, finishRevokeChan)
		for e := leaseHolderList.Front(); e != nil; e = e.Next() {
			leaseHolder := (e.Value).(LeaseRecord)
			go ss.revokeLease(leaseHolder.HostPort, args.Key, singleRevokeChan)
		}
		//wait until all lease holders reply or leases have expired
		<-finishRevokeChan

		//delete the lease list for the key
		delete(ss.leaseMap, args.Key)
	}

	//modify value
	ss.keyValueMap[args.Key] = args.Value

	//resume granting lease for that key
	lock.Unlock()

	reply.Status = storagerpc.OK
	return nil
}

func (ss *storageServer) AppendToList(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	if ss.isReady == false {
		reply.Status = storagerpc.NotReady
	} else {
		if ss.getServerID(args.Key) != ss.nodeID {
			reply.Status = storagerpc.WrongServer
			return nil
		}
		ss.serverMutex.Lock()
		if ss.userListMutex[args.Key] == nil {
			ss.userListMutex[args.Key] = &sync.Mutex{}
		}
		ss.serverMutex.Unlock()
		ss.userListMutex[args.Key].Lock()
		defer ss.userListMutex[args.Key].Unlock()
		_, exist := ss.userList[args.Key]
		if exist == false {
			ss.userList[args.Key] = list.New()
			ss.userList[args.Key].PushBack(args.Value)
			ss.userListLease[args.Key] = make(map[string]*lease)
			reply.Status = storagerpc.OK
		} else {

			// revoke all leases if already exist in server before
			err := ss.revokeLease(args.Key, ss.userListLease[args.Key])
			if err != nil {
				return err
			}
			ss.userListLease[args.Key] = make(map[string]*lease)

			exist = false
			for i := ss.userList[args.Key].Front(); i != nil; i = i.Next() {
				if i.Value.(string) == args.Value {
					exist = true
					break
				}
			}
			if exist == true {
				reply.Status = storagerpc.ItemExists
			} else {
				reply.Status = storagerpc.OK
				ss.userList[args.Key].PushBack(args.Value)
			}
		}
	}
	return nil
}

func (ss *storageServer) Put(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	if DBG {
		fmt.Println("Entered storage server put with key ", args.Key)
	}

	if !ss.inRange(libstore.StoreHash(args.Key)) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	// Leasing check
	ss.leaseLock.Lock()
	// If we have not seen this key before, initialize lease tracker
	_, ok := ss.pendingMap[args.Key]
	if !ok {
		ss.pendingMap[args.Key] = &TrackPending{pending: 0, pendingCh: make(chan chan int, 1)}
	}
	pendingModifications, _ := ss.pendingMap[args.Key]
	pendingModifications.pending++
	leaseHolders, ok := ss.leaseStore[args.Key]
	ss.leaseLock.Unlock()
	if pendingModifications.pending > 1 { // Block until it's our turn to modify key
		response := make(chan int)
		pendingModifications.pendingCh <- response
		<-response
	}
	if ok {
		ss.revokeLeases(leaseHolders, args.Key)
	}
	ss.dataLock.Lock()
	ss.dataStore[args.Key] = args.Value
	reply.Status = storagerpc.OK
	ss.dataLock.Unlock()
	pendingModifications.pending--
Loop:
	for {
		select {
		case ch := <-pendingModifications.pendingCh:
			ch <- 1
			break Loop
		default:
			break Loop
		}
	}
	return nil
}

func (ss *storageServer) Put(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	status, ok := ss.assertKeyAndServer(args.Key)
	if !ok {
		reply.Status = status
		return nil
	}

	ss.rwl.Lock()
	defer ss.rwl.Unlock()

	ss.waitForRevoking(args.Key)
	ss.revokeLease(args.Key)

	ss.store[args.Key] = args.Value
	reply.Status = storagerpc.OK
	return nil
}

func (ss *storageServer) RemoveFromList(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	key := args.Key

	if rightStorageServer(ss, key) == false {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	if _, found := ss.keyLocks[key]; found == false {
		ss.keyLocks[key] = make(chan int, 1)
	} else {
		<-ss.keyLocks[key]
	}

	if lst, found := ss.topMap[key]; found {
		if l, ok := lst.([]string); ok {

			for i := 0; i < len(l); i++ {
				if l[i] == args.Value {
					// found item in list, remove it and return
					reply.Status = storagerpc.OK
					ss.topMap[key] = append(l[:i], l[i+1:]...)
					ss.keyLocks[key] <- 1

					return nil
				}
			}
			// item was not in the list
			reply.Status = storagerpc.ItemNotFound
			ss.keyLocks[key] <- 1

			return nil
		} else {
			// list was corrupted, shouldn't happen
			ss.keyLocks[key] <- 1

			return errors.New("List to remove from is wrong type")
		}
	} else {
		reply.Status = storagerpc.KeyNotFound
		ss.keyLocks[key] <- 1

		return nil
	}
}

func (ss *storageServer) Put(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {

	key := args.Key
	if rightStorageServer(ss, key) == false {
		reply.Status = storagerpc.WrongServer

		return nil
	}

	if _, found := ss.keyLocks[key]; found == false {
		ss.keyLocks[key] = make(chan int, 1)
	} else {
		<-ss.keyLocks[key]
	}

	reply.Status = storagerpc.OK
	ss.topMap[key] = args.Value
	ss.keyLocks[key] <- 1
	return nil
}

func (pc *proxyCounter) RemoveFromList(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	if pc.override {
		reply.Status = pc.overrideStatus
		return pc.overrideErr
	}
	byteCount := len(args.Key) + len(args.Value)
	err := pc.srv.Call("StorageServer.RemoveFromList", args, reply)
	atomic.AddUint32(&pc.rpcCount, 1)
	atomic.AddUint32(&pc.byteCount, uint32(byteCount))
	return err
}

func (ss *storageServer) RemoveFromList(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	key := args.Key

	if !correctServer(ss, key) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	leases := getLeases(ss, key)
	if leases != nil {
		ss.editLock.Lock()
		revokeLease(ss, key)
		ss.editLock.Unlock()
	}

	ss.listMapLock.Lock()
	defer ss.listMapLock.Unlock()

	value := args.Value
	list, in := ss.listMap[key]

	if !in {
		reply.Status = storagerpc.ItemNotFound
	} else {
		index := -1
		for i := 0; i < len(list); i++ {
			if list[i] == value {
				index = i
				break
			}
		}
		if index == -1 {
			reply.Status = storagerpc.ItemNotFound
		} else {
			ss.listMap[key] = append(list[:index], list[(index+1):]...)
			reply.Status = storagerpc.OK
		}
	}
	return nil
}

// AppendToList retrieves the specified key from the data store and appends
// the specified value to its list. If the key does not fall within the
// receiving server's range, it should reply with status WrongServer. If
// the specified value is already contained in the list, it should reply
// with status ItemExists.
func (ss *storageServer) AppendToList(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	//	fmt.Println("AppendToList function called:", args);
	if FindPartitionNode(ss.ring, args.Key).NodeID != ss.nodeID {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	ss.sMutex.Lock()

	var list []string
	data, ok := ss.storage[args.Key]
	if !ok {
		//		fmt.Println("First time insert something to the list.. initialize it");
		list = make([]string, 0, 1)
		list = append(list, args.Value)

		ss.storage[args.Key] = &dataUnit{data: list, Set: nil}
		ss.sMutex.Unlock()
		reply.Status = storagerpc.OK
	} else {
		ss.sMutex.Unlock()
		data.dMutex.Lock()

		list = data.data.([]string)
		if ss.ListContains(list, args.Value) != -1 {
			reply.Status = storagerpc.ItemExists
		} else {
			// call the revoke call backs!
			ss.RevokeCallBacksForKey(args.Key, data)
			reply.Status = storagerpc.OK
			// apply changes
			list = append(list, args.Value)
			data.data = list
		}
		data.dMutex.Unlock()
	}

	return nil
}

func (ss *storageServer) AppendToList(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {

	key := args.Key

	if !correctServer(ss, key) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	leases := getLeases(ss, key)
	if leases != nil {
		ss.editLock.Lock()
		revokeLease(ss, key)
		ss.editLock.Unlock()
	}

	ss.listMapLock.Lock()
	defer ss.listMapLock.Unlock()

	value := args.Value
	list, in := ss.listMap[key]

	if !in {
		newList := []string{value}
		ss.listMap[key] = newList
	} else {
		for i := 0; i < len(list); i++ {
			if list[i] == value {
				reply.Status = storagerpc.ItemExists
				return nil
			}
		}
		ss.listMap[key] = append(list, value)
	}
	reply.Status = storagerpc.OK
	return nil
}

func (ss *storageServer) RemoveFromList(args *storagerpc.PutArgs, reply *storagerpc.PutReply) error {
	if !ss.IsKeyInRange(args.Key) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	/* Lock table for writing */
	ss.lockForTableLocks.Lock()
	lock, exist := ss.tableLocks[args.Key]
	if !exist {
		lock = new(sync.Mutex)
		ss.tableLocks[args.Key] = lock
	}
	ss.lockForTableLocks.Unlock()

	lock.Lock()
	defer lock.Unlock()

	ss.RevokeLease(args.Key)

	list := ss.listTable[args.Key]
	exist = false
	for i, item := range list {
		if item == args.Value {
			exist = true
			ss.listTable[args.Key] = append(list[:i], list[i+1:]...)
			break
		}
	}
	if !exist {
		reply.Status = storagerpc.ItemNotFound
	} else {
		reply.Status = storagerpc.OK
	}
	return nil
}